Action Service Accounts Proof of Concept

core/src/main/scala/org/broadinstitute/dsde/rawls/dataaccess/GoogleServicesDAO.scala

@@ -34,7 +34,8 @@ trait GoogleServicesDAO extends ErrorReportable {
   def updateBucketIam(bucketName: GcsBucketName,
                       policyGroupsByAccessLevel: Map[WorkspaceAccessLevel, WorkbenchEmail],
                       userProject: Option[GoogleProjectId] = None,
-                      iamPolicyVersion: Int = 1
+                      iamPolicyVersion: Int = 1,
+                      actionServiceAccountsByAction: Map[SamResourceAction, WorkbenchEmail] = Map.empty
   ): Future[Unit]
 
   // returns bucket and group information
@@ -44,7 +45,8 @@ trait GoogleServicesDAO extends ErrorReportable {
                      bucketName: GcsBucketName,
                      labels: Map[String, String],
                      requestContext: RawlsRequestContext,
-                     bucketLocation: Option[String]
+                     bucketLocation: Option[String],
+                     actionServiceAccountsByAction: Map[SamResourceAction, WorkbenchEmail] = Map.empty
   ): Future[GoogleWorkspaceInfo]
 
   def getGoogleProject(googleProject: GoogleProjectId): Future[Project]

core/src/main/scala/org/broadinstitute/dsde/rawls/dataaccess/HttpGoogleServicesDAO.scala

@@ -133,7 +133,8 @@ class HttpGoogleServicesDAO(val clientSecrets: GoogleClientSecrets,
   override def updateBucketIam(bucketName: GcsBucketName,
                                policyGroupsByAccessLevel: Map[WorkspaceAccessLevel, WorkbenchEmail],
                                userProject: Option[GoogleProjectId],
-                               iamPolicyVersion: Int = 1
+                               iamPolicyVersion: Int = 1,
+                               actionServiceAccountsByAction: Map[SamResourceAction, WorkbenchEmail] = Map.empty
   ): Future[Unit] = {
     // default object ACLs are no longer used. bucket only policy is enabled on buckets to ensure that objects
     // do not have separate permissions that deviate from the bucket-level permissions.
@@ -154,9 +155,19 @@ class HttpGoogleServicesDAO(val clientSecrets: GoogleClientSecrets,
       Read -> customTerraBucketReaderRole
     )
 
+    val samActionToStorageRole = Map(SamWorkspaceActions.write -> customTerraBucketWriterRole,
+                                     SamWorkspaceActions.read -> customTerraBucketReaderRole
+    )
+
     val roleToIdentities = policyGroupsByAccessLevel
       .map { case (access, policyEmail) => Identity.group(policyEmail.value) -> workspaceAccessToStorageRole(access) }
       .+(Identity.serviceAccount(clientEmail) -> StorageRole.StorageAdmin)
+      .++(
+        actionServiceAccountsByAction
+          .map { case (action, serviceAccountEmail) =>
+            Identity.serviceAccount(serviceAccountEmail.value) -> samActionToStorageRole(action)
+          }
+      )
       .groupBy(_._2)
       .view
       .mapValues(_.keys)
@@ -195,7 +206,8 @@ class HttpGoogleServicesDAO(val clientSecrets: GoogleClientSecrets,
                               bucketName: GcsBucketName,
                               labels: Map[String, String],
                               requestContext: RawlsRequestContext,
-                              bucketLocation: Option[String]
+                              bucketLocation: Option[String],
+                              actionServiceAccountsByAction: Map[SamResourceAction, WorkbenchEmail] = Map.empty
   ): Future[GoogleWorkspaceInfo] = {
     def insertInitialStorageLog: Future[Unit] = {
       implicit val service = GoogleInstrumentedService.Storage
@@ -258,7 +270,10 @@ class HttpGoogleServicesDAO(val clientSecrets: GoogleClientSecrets,
           }
       ) // ACL = None because bucket IAM will be set separately in updateBucketIam
       updateBucketIamFuture = traceFutureWithParent("updateBucketIam", requestContext)(_ =>
-        updateBucketIam(bucketName, policyGroupsByAccessLevel)
+        updateBucketIam(bucketName,
+                        policyGroupsByAccessLevel,
+                        actionServiceAccountsByAction = actionServiceAccountsByAction
+        )
       )
       insertInitialStorageLogFuture = traceFutureWithParent("insertInitialStorageLog", requestContext)(_ =>
         insertInitialStorageLog

core/src/main/scala/org/broadinstitute/dsde/rawls/dataaccess/HttpSamDAO.scala

@@ -340,6 +340,24 @@ class HttpSamDAO(baseSamServiceURL: String, rawlsCredential: RawlsCredential, ti
       callback.future.map(_.booleanValue())
     }
 
+  override def getActionServiceAccount(googleProject: GoogleProjectId,
+                                       resourceTypeName: SamResourceTypeName,
+                                       resourceId: String,
+                                       action: SamResourceAction,
+                                       ctx: RawlsRequestContext
+  ): Future[WorkbenchEmail] = retry(when401or5xx) { () =>
+    val callback = new SamApiCallback[java.lang.String]("getActionServiceAccount")
+
+    googleApi(ctx).getActionServiceAccount(googleProject.value,
+                                           resourceTypeName.value,
+                                           resourceId,
+                                           action.value,
+                                           callback
+    )
+
+    callback.future.map(WorkbenchEmail)
+  }
+
   override def getPolicy(resourceTypeName: SamResourceTypeName,
                          resourceId: String,
                          policyName: SamResourcePolicyName,

core/src/main/scala/org/broadinstitute/dsde/rawls/dataaccess/SamDAO.scala

@@ -62,6 +62,13 @@ trait SamDAO {
                     cts: RawlsRequestContext
   ): Future[Boolean]
 
+  def getActionServiceAccount(googleProject: GoogleProjectId,
+                              resourceTypeName: SamResourceTypeName,
+                              resourceId: String,
+                              action: SamResourceAction,
+                              ctx: RawlsRequestContext
+  ): Future[WorkbenchEmail]
+
   def getPolicy(resourceTypeName: SamResourceTypeName,
                 resourceId: String,
                 policyName: SamResourcePolicyName,

core/src/main/scala/org/broadinstitute/dsde/rawls/methods/MethodConfigurationUtils.scala

@@ -21,7 +21,7 @@ object MethodConfigurationUtils {
       case Success(None) =>
         throw new RawlsExceptionWithErrorReport(
           errorReport = ErrorReport(StatusCodes.NotFound,
-            s"Cannot get ${methodConfig.methodRepoMethod.methodUri} from method repo."
+                                    s"Cannot get ${methodConfig.methodRepoMethod.methodUri} from method repo."
           )
         )
       case Success(Some(wdl)) =>
@@ -34,8 +34,8 @@ object MethodConfigurationUtils {
       case Failure(throwable) =>
         throw new RawlsExceptionWithErrorReport(
           errorReport = ErrorReport(StatusCodes.BadGateway,
-            s"Unable to query the method repo.",
-            methodRepoDAO.toErrorReport(throwable)
+                                    s"Unable to query the method repo.",
+                                    methodRepoDAO.toErrorReport(throwable)
           )
         )
     }

core/src/main/scala/org/broadinstitute/dsde/rawls/workspace/WorkspaceService.scala

@@ -2094,6 +2094,7 @@ class WorkspaceService(
   def setupGoogleProjectIam(googleProjectId: GoogleProjectId,
                             policyEmailsByName: Map[SamResourcePolicyName, WorkbenchEmail],
                             billingProjectOwnerPolicyEmail: WorkbenchEmail,
+                            actionServiceAccountsByAction: Map[SamResourceAction, WorkbenchEmail],
                             parentContext: RawlsRequestContext = ctx
   ): Future[Unit] =
     traceFutureWithParent("updateGoogleProjectIam", parentContext) { _ =>
@@ -2111,27 +2112,43 @@ class WorkspaceService(
 
       // todo: update this line as part of https://broadworkbench.atlassian.net/browse/CA-1220
       // This is done sequentially intentionally in order to avoid conflict exceptions as a result of concurrent IAM updates.
-      List(
-        billingProjectOwnerPolicyEmail -> Set(terraBillingProjectOwnerRole,
-                                              terraWorkspaceCanComputeRole,
-                                              terraWorkspaceNextflowRole
-        ),
-        policyEmailsByName(SamWorkspacePolicyNames.owner) -> Set(terraWorkspaceCanComputeRole,
-                                                                 terraWorkspaceNextflowRole
-        ),
-        policyEmailsByName(SamWorkspacePolicyNames.canCompute) -> Set(terraWorkspaceCanComputeRole,
-                                                                      terraWorkspaceNextflowRole
+      for {
+        _ <- List(
+          billingProjectOwnerPolicyEmail -> Set(terraBillingProjectOwnerRole,
+                                                terraWorkspaceCanComputeRole,
+                                                terraWorkspaceNextflowRole
+          ),
+          policyEmailsByName(SamWorkspacePolicyNames.owner) -> Set(terraWorkspaceCanComputeRole,
+                                                                   terraWorkspaceNextflowRole
+          ),
+          policyEmailsByName(SamWorkspacePolicyNames.canCompute) -> Set(terraWorkspaceCanComputeRole,
+                                                                        terraWorkspaceNextflowRole
+          )
         )
-      )
-        .traverse_ { case (email, roles) =>
-          googleIamDao.addRoles(
-            GoogleProject(googleProjectId.value),
-            email,
-            IamMemberTypes.Group,
-            roles,
-            retryIfGroupDoesNotExist = true
+          .traverse_ { case (email, roles) =>
+            googleIamDao.addRoles(
+              GoogleProject(googleProjectId.value),
+              email,
+              IamMemberTypes.Group,
+              roles,
+              retryIfGroupDoesNotExist = true
+            )
+          }
+        _ <- List(
+          actionServiceAccountsByAction(SamWorkspaceActions.compute) -> Set(terraWorkspaceCanComputeRole,
+                                                                            terraWorkspaceNextflowRole
           )
-        }
+        )
+          .traverse_ { case (email, roles) =>
+            googleIamDao.addRoles(
+              GoogleProject(googleProjectId.value),
+              email,
+              IamMemberTypes.ServiceAccount,
+              roles
+            )
+          }
+      } yield ()
+
     }
 
   /**
@@ -2316,6 +2333,21 @@ class WorkspaceService(
       }
     )
 
+  private def createActionServiceAccountsInSam(workspaceId: String,
+                                               googleProjectId: GoogleProjectId,
+                                               actions: Set[SamResourceAction],
+                                               parentContext: RawlsRequestContext
+  ): Future[Map[SamResourceAction, WorkbenchEmail]] =
+    traceFutureWithParent("createActionServiceAccountsInSam", parentContext)(_ =>
+      Future
+        .traverse(actions) { action =>
+          samDAO
+            .getActionServiceAccount(googleProjectId, SamResourceTypeNames.workspace, workspaceId, action, ctx)
+            .map(action -> _)
+        }
+        .map(_.toMap)
+    )
+
   private def createWorkspaceInDatabase(workspaceId: String,
                                         workspaceRequest: WorkspaceRequest,
                                         bucketName: String,
@@ -2433,7 +2465,9 @@ class WorkspaceService(
           syncPolicies(workspaceId, policyEmailsByName, workspaceRequest, parentContext)
         ).sequence_
       }
-      (googleProjectId, googleProjectNumber) <- traceDBIOWithParent("setupGoogleProject", parentContext) { span =>
+      (googleProjectId, googleProjectNumber, actionServiceAccountsByAction) <- traceDBIOWithParent("setupGoogleProject",
+                                                                                                   parentContext
+      ) { span =>
         DBIO.from(
           for {
             (googleProjectId, googleProjectNumber) <- createGoogleProject(billingProject,
@@ -2442,8 +2476,22 @@ class WorkspaceService(
             )
             _ <- setProjectBillingAccount(googleProjectId, billingProject, billingAccount, workspaceId, span)
             _ <- renameAndLabelProject(googleProjectId, workspaceId, workspaceName, span)
-            _ <- setupGoogleProjectIam(googleProjectId, policyEmailsByName, billingProjectOwnerPolicyEmail, span)
-          } yield (googleProjectId, googleProjectNumber)
+            actionsForServiceAccounts = Set(SamWorkspaceActions.read,
+                                            SamWorkspaceActions.write,
+                                            SamWorkspaceActions.compute
+            )
+            actionServiceAccountsByAction <- createActionServiceAccountsInSam(workspaceId,
+                                                                              googleProjectId,
+                                                                              actionsForServiceAccounts,
+                                                                              parentContext
+            )
+            _ <- setupGoogleProjectIam(googleProjectId,
+                                       policyEmailsByName,
+                                       billingProjectOwnerPolicyEmail,
+                                       actionServiceAccountsByAction,
+                                       span
+            )
+          } yield (googleProjectId, googleProjectNumber, actionServiceAccountsByAction)
         )
       }
       savedWorkspace <- traceDBIOWithParent("createWorkspaceInDatabase", parentContext)(span =>
@@ -2516,7 +2564,8 @@ class WorkspaceService(
             GcsBucketName(bucketName),
             getLabels(workspaceRequest.authorizationDomain.getOrElse(Set.empty).toList),
             childContext,
-            workspaceBucketLocation
+            workspaceBucketLocation,
+            actionServiceAccountsByAction
           )
         )
       )

core/src/test/scala/org/broadinstitute/dsde/rawls/dataaccess/MockGoogleServicesDAO.scala

@@ -91,7 +91,8 @@ class MockGoogleServicesDAO(groupsPrefix: String,
   override def updateBucketIam(bucketName: GcsBucketName,
                                policyGroupsByAccessLevel: Map[WorkspaceAccessLevel, WorkbenchEmail],
                                userProject: Option[GoogleProjectId],
-                               iamPolicyVersion: Int = 1
+                               iamPolicyVersion: Int = 1,
+                               actionServiceAccountsByAction: Map[SamResourceAction, WorkbenchEmail] = Map.empty
   ): Future[Unit] =
     Future.unit
 
@@ -101,7 +102,8 @@ class MockGoogleServicesDAO(groupsPrefix: String,
                               bucketName: GcsBucketName,
                               labels: Map[String, String],
                               requestContext: RawlsRequestContext,
-                              bucketLocation: Option[String]
+                              bucketLocation: Option[String],
+                              actionServiceAccountsByAction: Map[SamResourceAction, WorkbenchEmail] = Map.empty
   ): Future[GoogleWorkspaceInfo] = {
 
     val googleWorkspaceInfo: GoogleWorkspaceInfo = GoogleWorkspaceInfo(bucketName.value, policyGroupsByAccessLevel)

core/src/test/scala/org/broadinstitute/dsde/rawls/methods/MethodConfigurationUtilsSpec.scala

@@ -4,15 +4,27 @@ import akka.actor.ActorSystem
 import akka.http.scaladsl.model.StatusCodes
 import akka.http.scaladsl.model.headers.OAuth2BearerToken
 import com.typesafe.config.ConfigFactory
-import org.broadinstitute.dsde.rawls.{RawlsException, RawlsExceptionWithErrorReport, TestExecutionContext, jobexec}
+import org.broadinstitute.dsde.rawls.{jobexec, RawlsException, RawlsExceptionWithErrorReport, TestExecutionContext}
 import org.broadinstitute.dsde.rawls.config.{MethodRepoConfig, WDLParserConfig}
 import org.broadinstitute.dsde.rawls.dataaccess.{HttpMethodRepoDAO, MockCromwellSwaggerClient}
 import org.broadinstitute.dsde.rawls.dataaccess.slick.TestDriverComponent
 import org.broadinstitute.dsde.rawls.jobexec.MethodConfigResolver
 import org.broadinstitute.dsde.rawls.jobexec.MethodConfigResolver.GatherInputsResult
 import org.broadinstitute.dsde.rawls.jobexec.wdlparsing.CachingWDLParser
 import org.broadinstitute.dsde.rawls.mock.RemoteServicesMockServer
-import org.broadinstitute.dsde.rawls.model.{Agora, AgoraMethod, Dockstore, ErrorReport, ErrorReportSource, MethodConfiguration, RawlsRequestContext, RawlsUser, RawlsUserEmail, RawlsUserSubjectId, UserInfo}
+import org.broadinstitute.dsde.rawls.model.{
+  Agora,
+  AgoraMethod,
+  Dockstore,
+  ErrorReport,
+  ErrorReportSource,
+  MethodConfiguration,
+  RawlsRequestContext,
+  RawlsUser,
+  RawlsUserEmail,
+  RawlsUserSubjectId,
+  UserInfo
+}
 import org.mockito.ArgumentMatchers.any
 import org.mockito.Mockito.when
 import org.scalatest.flatspec.AnyFlatSpec
@@ -30,34 +42,43 @@ class MethodConfigurationUtilsSpec extends AnyFlatSpec with Matchers with TestDr
 
   val mockServer: RemoteServicesMockServer = RemoteServicesMockServer()
   val methodRepoDAO: HttpMethodRepoDAO = mock[HttpMethodRepoDAO]
-  val ctx: RawlsRequestContext = RawlsRequestContext(UserInfo(testData.userProjectOwner.userEmail, OAuth2BearerToken("foo"), 0, testData.userProjectOwner.userSubjectId))
+  val ctx: RawlsRequestContext = RawlsRequestContext(
+    UserInfo(testData.userProjectOwner.userEmail, OAuth2BearerToken("foo"), 0, testData.userProjectOwner.userSubjectId)
+  )
 
   val agoraMethodConf: MethodConfiguration = MethodConfiguration("dsde",
-    "no_input",
-    Some("Sample"),
-    None,
-    Map.empty,
-    Map.empty,
-    AgoraMethod("dsde", "no_input", 1))
+                                                                 "no_input",
+                                                                 Some("Sample"),
+                                                                 None,
+                                                                 Map.empty,
+                                                                 Map.empty,
+                                                                 AgoraMethod("dsde", "no_input", 1)
+  )
 
   behavior of "MethodConfigurationUtils"
 
   it should "return results when method when found" in {
     when(methodRepoDAO.getMethod(any(), any())).thenReturn(Future.successful(Option(meth1WDL)))
 
-    val future = MethodConfigurationUtils.gatherMethodConfigInputs(ctx, methodRepoDAO, agoraMethodConf, methodConfigResolver)
+    val future =
+      MethodConfigurationUtils.gatherMethodConfigInputs(ctx, methodRepoDAO, agoraMethodConf, methodConfigResolver)
     Await.ready(future, 30.seconds)
     val Success(result) = future.value.get
 
     // verify that it returns GatherInputsResult
-    result shouldBe a [GatherInputsResult]
+    result shouldBe a[GatherInputsResult]
     result.missingInputs shouldBe Set("meth1.method1.i1")
   }
 
   it should "return 404 if method is not found" in {
     when(methodRepoDAO.getMethod(any(), any())).thenReturn(Future.successful(None))
 
-    val exception = intercept[RawlsExceptionWithErrorReport](Await.result(MethodConfigurationUtils.gatherMethodConfigInputs(ctx, methodRepoDAO, agoraMethodConf, methodConfigResolver), 30.seconds))
+    val exception = intercept[RawlsExceptionWithErrorReport](
+      Await.result(
+        MethodConfigurationUtils.gatherMethodConfigInputs(ctx, methodRepoDAO, agoraMethodConf, methodConfigResolver),
+        30.seconds
+      )
+    )
 
     // assert that if method is not found it returns 404
     exception shouldBe a[RawlsExceptionWithErrorReport]
@@ -66,11 +87,17 @@ class MethodConfigurationUtilsSpec extends AnyFlatSpec with Matchers with TestDr
   }
 
   it should "return 502 when something unexpected happens" in {
-    when(methodRepoDAO.getMethod(any(), any())).thenReturn(Future.failed(new RawlsException("exception thrown for testing purposes")))
+    when(methodRepoDAO.getMethod(any(), any()))
+      .thenReturn(Future.failed(new RawlsException("exception thrown for testing purposes")))
     when(methodRepoDAO.errorReportSource).thenReturn(ErrorReportSource("agora"))
     when(methodRepoDAO.toErrorReport(any())).thenCallRealMethod()
 
-    val exception = intercept[RawlsExceptionWithErrorReport](Await.result(MethodConfigurationUtils.gatherMethodConfigInputs(ctx, methodRepoDAO, agoraMethodConf, methodConfigResolver), 30.seconds))
+    val exception = intercept[RawlsExceptionWithErrorReport](
+      Await.result(
+        MethodConfigurationUtils.gatherMethodConfigInputs(ctx, methodRepoDAO, agoraMethodConf, methodConfigResolver),
+        30.seconds
+      )
+    )
 
     // assert that when Future fails, a 502 is returned with wrapped exception
     exception shouldBe a[RawlsExceptionWithErrorReport]