[CORE-53] Fix storage request errors on requester pays workspaces (#3097

)

core/src/main/resources/swagger/api-docs.yaml

@@ -3194,6 +3194,7 @@ paths:
       parameters:
         - $ref: '#/components/parameters/workspaceNamespacePathParam'
         - $ref: '#/components/parameters/workspaceNamePathParam'
+        - $ref: '#/components/parameters/userProjectQueryParam'
       responses:
         200:
           description: Successful Request
@@ -3386,6 +3387,7 @@ paths:
             type: array
             items:
               type: string
+        - $ref: '#/components/parameters/userProjectQueryParam'
       responses:
         200:
           description: Successful Request
@@ -3428,6 +3430,7 @@ paths:
             type: array
             items:
               type: string
+        - $ref: '#/components/parameters/userProjectQueryParam'
       responses:
         200:
           description: Successful Request
@@ -6325,11 +6328,15 @@ components:
     WorkspaceBucketOptions:
       required:
         - requesterPays
+        - location
       type: object
       properties:
         requesterPays:
           type: boolean
           description: Whether the bucket is requester pays
+        location:
+          type: string
+          description: The region the bucket is in
       description: Extra information about a GCS bucket attached to a workspace
     WorkspaceAccessLevel:
       type: string
@@ -7213,6 +7220,14 @@ components:
       required: true
       schema:
         type: string
+    userProjectQueryParam:
+      name: userProject
+      in: query
+      description: |
+        When specified, bill google storage requests to this project
+      required: false
+      schema:
+        type: string
     workbenchRolePathParam:
       name: workbenchRole
       in: path

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

@@ -1100,9 +1100,11 @@ class HttpGoogleServicesDAO(val clientSecrets: GoogleClientSecrets,
         billing <- Option(bucketDetails.getBilling)
         rp <- Option(billing.getRequesterPays)
       } yield rp.booleanValue()
+      val location = Option(bucketDetails.getLocation)
 
       WorkspaceBucketOptions(
-        requesterPays = requesterPays.getOrElse(false)
+        requesterPays = requesterPays.getOrElse(false),
+        location = location.getOrElse("")
       )
     }
   }

core/src/main/scala/org/broadinstitute/dsde/rawls/dataaccess/slick/WorkspaceComponent.scala

@@ -365,6 +365,9 @@ trait WorkspaceComponent {
       _.getOrElse(throw new RawlsException(s"""No workspace found matching id "$workspaceId"."""))
     }
 
+    def findByGoogleProjectId(googleProjectId: GoogleProjectId): ReadAction[Option[Workspace]] =
+      loadWorkspace(findByGoogleProjectIdQuery(googleProjectId))
+
     def listByIds(workspaceIds: Seq[UUID],
                   attributeSpecs: Option[WorkspaceAttributeSpecs] = None
     ): ReadAction[Seq[Workspace]] =
@@ -596,6 +599,9 @@ trait WorkspaceComponent {
     def findByGoogleProjectNumbersQuery(googleProjectNumbers: Seq[String]): WorkspaceQueryType =
       filter(w => w.googleProjectNumber.map(_.inSetBind(googleProjectNumbers)))
 
+    def findByGoogleProjectIdQuery(googleProjectId: GoogleProjectId): WorkspaceQueryType =
+      workspaceQuery.withGoogleProjectId(googleProjectId)
+
     private def loadWorkspace(lookup: WorkspaceQueryType,
                               attributeSpecs: Option[WorkspaceAttributeSpecs] = None
     ): ReadAction[Option[Workspace]] =

core/src/main/scala/org/broadinstitute/dsde/rawls/dataaccess/slick/WorkspaceSettingComponent.scala

@@ -128,5 +128,16 @@ trait WorkspaceSettingComponent {
     ): ReadAction[List[WorkspaceSetting]] =
       filter(rec => rec.workspaceId === workspaceId && rec.status === status.toString).result
         .map(_.map(WorkspaceSettingRecord.toWorkspaceSetting).toList)
+
+    def getAppliedSettingForWorkspaceByType(workspaceId: UUID,
+                                            settingType: WorkspaceSettingType
+    ): ReadAction[Option[WorkspaceSetting]] =
+      uniqueResult(
+        filter(rec =>
+          rec.workspaceId === workspaceId
+            && rec.status === WorkspaceSettingRecord.SettingStatus.Applied.toString
+            && rec.settingType === settingType.toString
+        ).take(1).result.map(_.map(WorkspaceSettingRecord.toWorkspaceSetting).toList)
+      )
   }
 }

core/src/main/scala/org/broadinstitute/dsde/rawls/webservice/WorkspaceApiService.scala

@@ -67,13 +67,16 @@ trait WorkspaceApiService extends UserInfoDirectives {
         } ~
         path("workspaces" / "id" / Segment) { workspaceId =>
           get {
-            parameterSeq { allParams =>
-              complete {
-                workspaceServiceConstructor(ctx).getWorkspaceById(workspaceId,
-                                                                  WorkspaceFieldSpecs.fromQueryParams(allParams,
-                                                                                                      "fields"
-                                                                  )
-                )
+            parameters("userProject".optional) { userProject =>
+              parameterSeq { allParams =>
+                complete {
+                  workspaceServiceConstructor(ctx).getWorkspaceById(workspaceId,
+                                                                    WorkspaceFieldSpecs.fromQueryParams(allParams,
+                                                                                                        "fields"
+                                                                    ),
+                                                                    userProject.map(GoogleProjectId)
+                  )
+                }
               }
             }
           }
@@ -89,11 +92,16 @@ trait WorkspaceApiService extends UserInfoDirectives {
             }
           } ~
             get {
-              parameterSeq { allParams =>
-                complete {
-                  workspaceServiceConstructor(ctx).getWorkspace(WorkspaceName(workspaceNamespace, workspaceName),
-                                                                WorkspaceFieldSpecs.fromQueryParams(allParams, "fields")
-                  )
+              parameters("userProject".optional) { userProject =>
+                parameterSeq { allParams =>
+                  complete {
+                    workspaceServiceConstructor(ctx).getWorkspace(WorkspaceName(workspaceNamespace, workspaceName),
+                                                                  WorkspaceFieldSpecs.fromQueryParams(allParams,
+                                                                                                      "fields"
+                                                                  ),
+                                                                  userProject.map(GoogleProjectId)
+                    )
+                  }
                 }
               }
             } ~
@@ -116,8 +124,13 @@ trait WorkspaceApiService extends UserInfoDirectives {
         } ~
         path("workspaces" / Segment / Segment / "bucketOptions") { (workspaceNamespace, workspaceName) =>
           get {
-            complete {
-              workspaceServiceConstructor(ctx).getBucketOptions(WorkspaceName(workspaceNamespace, workspaceName))
+            parameters("userProject".optional) { userProject =>
+              complete {
+                workspaceServiceConstructor(ctx).getBucketOptions(
+                  WorkspaceName(workspaceNamespace, workspaceName),
+                  userProject.map(GoogleProjectId)
+                )
+              }
             }
           }
         } ~

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

@@ -7,6 +7,7 @@ import org.broadinstitute.dsde.rawls.dataaccess.slick.PendingBucketDeletionRecor
 import org.broadinstitute.dsde.rawls.model.Attributable.AttributeMap
 import org.broadinstitute.dsde.rawls.model.{
   ErrorReport,
+  GoogleProjectId,
   PendingCloneWorkspaceFileTransfer,
   RawlsBillingProjectName,
   RawlsRequestContext,
@@ -52,6 +53,11 @@ class WorkspaceRepository(dataSource: SlickDataSource) {
     _.workspaceQuery.getV2WorkspaceId(workspaceName)
   }
 
+  def getWorkspaceByGoogleProject(googleProjectId: GoogleProjectId): Future[Option[Workspace]] =
+    dataSource.inTransaction { access =>
+      access.workspaceQuery.findByGoogleProjectId(googleProjectId)
+    }
+
   def listWorkspacesByIds(workspaceIds: Seq[UUID],
                           attributeSpecs: Option[WorkspaceAttributeSpecs] = None
   ): Future[Seq[Workspace]] = dataSource.inTransaction {

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

@@ -24,6 +24,7 @@ import org.broadinstitute.dsde.rawls.metrics.{MetricsHelper, RawlsInstrumented}
 import org.broadinstitute.dsde.rawls.model.AttributeUpdateOperations._
 import org.broadinstitute.dsde.rawls.model.WorkspaceAccessLevels._
 import org.broadinstitute.dsde.rawls.model.WorkspaceJsonSupport._
+import org.broadinstitute.dsde.rawls.model.WorkspaceSettingTypes.GcpBucketRequesterPays
 import org.broadinstitute.dsde.rawls.model.WorkspaceState.WorkspaceState
 import org.broadinstitute.dsde.rawls.model.WorkspaceType.WorkspaceType
 import org.broadinstitute.dsde.rawls.model._
@@ -118,7 +119,8 @@ object WorkspaceService {
       (context: RawlsRequestContext) => fastPassServiceConstructor(context, dataSource),
       new WorkspaceRepository(dataSource),
       new BillingRepository(dataSource),
-      new SubmissionsRepository(dataSource, config.trackDetailedSubmissionMetrics, workbenchMetricBaseName)
+      new SubmissionsRepository(dataSource, config.trackDetailedSubmissionMetrics, workbenchMetricBaseName),
+      new WorkspaceSettingRepository(dataSource)
     )
 
   val SECURITY_LABEL_KEY: String = "security"
@@ -166,7 +168,8 @@ class WorkspaceService(
   val fastPassServiceConstructor: RawlsRequestContext => FastPassService,
   val workspaceRepository: WorkspaceRepository,
   val billingRepository: BillingRepository,
-  val submissionsRepository: SubmissionsRepository
+  val submissionsRepository: SubmissionsRepository,
+  val workspaceSettingsRepository: WorkspaceSettingRepository
 )(implicit protected val executionContext: ExecutionContext)
     extends LazyLogging
     with LibraryPermissionsSupport
@@ -229,33 +232,42 @@ class WorkspaceService(
     } yield workspace
   }
 
-  def getWorkspace(workspaceName: WorkspaceName, params: WorkspaceFieldSpecs): Future[JsObject] = {
+  def getWorkspace(workspaceName: WorkspaceName,
+                   params: WorkspaceFieldSpecs,
+                   userProject: Option[GoogleProjectId] = None
+  ): Future[JsObject] = {
     val options = processOptions(params)
     traceFutureWithParent("getV2WorkspaceContextAndPermissions", ctx)(_ =>
       for {
         workspace <-
           getV2WorkspaceContextAndPermissions(workspaceName, SamWorkspaceActions.read, Option(options.attrSpecs))
-        workspaceResponse <- getWorkspaceDetails(workspace, options)
+        workspaceResponse <- getWorkspaceDetails(workspace, options, userProject)
       } yield
       // post-process JSON to remove calculated-but-undesired keys
       deepFilterJsObject(workspaceResponse.toJson.asJsObject, options.options)
     )
   }
 
-  def getWorkspaceById(workspaceId: String, params: WorkspaceFieldSpecs): Future[JsObject] = {
+  def getWorkspaceById(workspaceId: String,
+                       params: WorkspaceFieldSpecs,
+                       userProject: Option[GoogleProjectId] = None
+  ): Future[JsObject] = {
     val options = processOptions(params)
     traceFutureWithParent("getV2WorkspaceContextAndPermissions", ctx)(_ =>
       for {
         workspace <-
           getV2WorkspaceContextAndPermissionsById(workspaceId, SamWorkspaceActions.read, Option(options.attrSpecs))
-        workspaceResponse <- getWorkspaceDetails(workspace, options)
+        workspaceResponse <- getWorkspaceDetails(workspace, options, userProject)
       } yield
       // post-process JSON to remove calculated-but-undesired keys
       deepFilterJsObject(workspaceResponse.toJson.asJsObject, options.options)
     )
   }
 
-  def getWorkspaceDetails(workspace: Workspace, options: QueryOptions): Future[WorkspaceResponse] = {
+  def getWorkspaceDetails(workspace: Workspace,
+                          options: QueryOptions,
+                          userProject: Option[GoogleProjectId] = None
+  ): Future[WorkspaceResponse] = {
     val workspaceId = workspace.workspaceId
     /*
     If we're looking to improve performance, we could potentially use this, instead trying to run futures in parallel:
@@ -311,9 +323,13 @@ class WorkspaceService(
           case _ => samDAO.userHasAction(SamResourceTypeNames.workspace, workspaceId, SamWorkspaceActions.compute, ctx)
         }
       }
+
       bucketDetails: Option[WorkspaceBucketOptions] <- wsmContext.googleProjectId match {
-        case None     => Future.successful(None)
-        case Some(id) => options.anyPresentFuture("bucketOptions")(gcsDAO.getBucketDetails(workspace.bucketName, id))
+        case None => Future.successful(None)
+        case Some(_) =>
+          options.anyPresentFuture("bucketOptions")(
+            getBucketOptions(WorkspaceName(workspace.namespace, workspace.name), userProject)
+          )
       }
     } yield WorkspaceResponse(
       options.anyPresent("accessLevel")(accessLevel),
@@ -972,7 +988,9 @@ class WorkspaceService(
         (aclChanges ++ existingAcls.filter(existingAcl => emailsBeingChanged.contains(existingAcl.email.toLowerCase)))
           .flatMap(aclUpdateToPolicies)
 
-      if (changingPolicies.exists(policy => !callingUserActions.contains(SamWorkspaceActions.sharePolicy(policy.value)))) {
+      if (
+        changingPolicies.exists(policy => !callingUserActions.contains(SamWorkspaceActions.sharePolicy(policy.value)))
+      ) {
         throw new InvalidWorkspaceAclUpdateException(
           ErrorReport(StatusCodes.BadRequest, "you do not have sufficient permissions to make these changes")
         )
@@ -1428,9 +1446,43 @@ class WorkspaceService(
       case Some(workspace) => op(workspace)
     }
 
-  def getBucketOptions(workspaceName: WorkspaceName): Future[WorkspaceBucketOptions] = for {
+  def getBucketOptions(workspaceName: WorkspaceName,
+                       userProject: Option[GoogleProjectId] = None
+  ): Future[WorkspaceBucketOptions] = for {
     workspaceContext <- getV2WorkspaceContextAndPermissions(workspaceName, SamWorkspaceActions.read)
-    options <- gcsDAO.getBucketDetails(workspaceContext.bucketName, workspaceContext.googleProjectId)
+    isWriter <- samDAO.userHasAction(SamResourceTypeNames.workspace,
+                                     workspaceContext.workspaceId,
+                                     SamWorkspaceActions.write,
+                                     ctx
+    )
+    requesterPays <- workspaceSettingsRepository
+      .getWorkspaceSettingOfType(workspaceContext.workspaceIdAsUUID, GcpBucketRequesterPays)
+      .map {
+        case Some(GcpBucketRequesterPaysSetting(config)) => config.enabled
+        case _                                           => false
+      }
+    userProjectWorkspace <- userProject.flatTraverse(workspaceRepository.getWorkspaceByGoogleProject)
+    isUserProjectWriter <- userProjectWorkspace.traverse(ws =>
+      samDAO.userHasAction(
+        SamResourceTypeNames.workspace,
+        ws.workspaceId,
+        SamWorkspaceActions.write,
+        ctx
+      )
+    )
+    _ = if (!isUserProjectWriter.getOrElse(true)) {
+      throw new RawlsExceptionWithErrorReport(
+        ErrorReport(StatusCodes.BadRequest, "User must have write access to user project")
+      )
+    }
+    _ = if (requesterPays && !isWriter && userProjectWorkspace.isEmpty) {
+      throw new RawlsExceptionWithErrorReport(
+        ErrorReport(StatusCodes.BadRequest, "Readers must provide a user project to bill on requester pays workspaces")
+      )
+    }
+    options <- gcsDAO.getBucketDetails(workspaceContext.bucketName,
+                                       userProject.getOrElse(workspaceContext.googleProjectId)
+    )
   } yield options
 
   def getBucketUsage(workspaceName: WorkspaceName): Future[BucketUsageResponse] = (for {

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

@@ -27,6 +27,14 @@ class WorkspaceSettingRepository(dataSource: SlickDataSource) {
       )
     }
 
+  // Return applied setting for a workspace of a certain setting type. Deleted and pending settings are not returned.
+  def getWorkspaceSettingOfType(workspaceId: UUID,
+                                settingType: WorkspaceSettingType
+  ): Future[Option[WorkspaceSetting]] =
+    dataSource.inTransaction { access =>
+      access.workspaceSettingQuery.getAppliedSettingForWorkspaceByType(workspaceId, settingType)
+    }
+
   // Create new settings for a workspace as pending. If there are any existing pending settings, throw an exception.
   def createWorkspaceSettingsRecords(workspaceId: UUID,
                                      workspaceSettings: List[WorkspaceSetting],

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

@@ -43,6 +43,8 @@ class MockGoogleServicesDAO(groupsPrefix: String,
 
   val mockJobIds = Seq("operations/dummy-job-id", "projects/dummy-project/operations/dummy-job-id")
 
+  val bucketLocation = "us-central1"
+
   override def listBillingAccounts(userInfo: UserInfo,
                                    firecloudHasAccess: Option[Boolean] = None
   ): Future[Seq[RawlsBillingAccount]] = {
@@ -230,7 +232,7 @@ class MockGoogleServicesDAO(groupsPrefix: String,
     Future.successful(true)
 
   override def getBucketDetails(bucket: String, project: GoogleProjectId): Future[WorkspaceBucketOptions] =
-    Future.successful(WorkspaceBucketOptions(false))
+    Future.successful(WorkspaceBucketOptions(false, bucketLocation))
 
   protected def updatePolicyBindings(
     googleProject: GoogleProjectId

core/src/test/scala/org/broadinstitute/dsde/rawls/webservice/PetSASpec.scala

@@ -106,7 +106,7 @@ class PetSASpec extends ApiServiceSpec {
                                                      Some(WorkspaceCloudPlatform.Gcp)
             ),
             Option(WorkspaceSubmissionStats(None, None, 0)),
-            Option(WorkspaceBucketOptions(false)),
+            Option(WorkspaceBucketOptions(false, services.gcsDAO.bucketLocation)),
             Option(Set.empty),
             None,
             None
@@ -155,7 +155,7 @@ class PetSASpec extends ApiServiceSpec {
                                                      Some(WorkspaceCloudPlatform.Gcp)
             ),
             Option(WorkspaceSubmissionStats(None, None, 0)),
-            Option(WorkspaceBucketOptions(false)),
+            Option(WorkspaceBucketOptions(false, services.gcsDAO.bucketLocation)),
             Option(Set.empty),
             None,
             None

core/src/test/scala/org/broadinstitute/dsde/rawls/webservice/WorkspaceApiGetOptionsSpec.scala

@@ -256,7 +256,7 @@ class WorkspaceApiGetOptionsSpec extends ApiServiceSpec {
                                              Some(WorkspaceCloudPlatform.Gcp)
     ),
     Option(WorkspaceSubmissionStats(Option(testDate), Option(testDate), 2)),
-    Option(WorkspaceBucketOptions(false)),
+    Option(WorkspaceBucketOptions(false, "us-central1")),
     Option(Set.empty),
     None,
     Some(List.empty)