diff --git a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResource.scala b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResource.scala index 93a601d0de0..78ec3f6cff6 100644 --- a/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResource.scala +++ b/amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResource.scala @@ -35,10 +35,11 @@ import org.apache.texera.amber.engine.common.storage.SequentialRecordStorage import org.apache.texera.amber.util.JSONUtils.objectMapper import org.apache.texera.amber.util.serde.GlobalPortIdentitySerde.SerdeOps import org.apache.texera.auth.{JwtParser, SessionUser} +import org.apache.texera.auth.util.ComputingUnitAccess import org.apache.texera.dao.SqlServer import org.apache.texera.dao.SqlServer.withTransaction import org.apache.texera.dao.jooq.generated.Tables._ -import org.apache.texera.dao.jooq.generated.enums.UserRoleEnum +import org.apache.texera.dao.jooq.generated.enums.{PrivilegeEnum, UserRoleEnum} import org.apache.texera.dao.jooq.generated.tables.daos.WorkflowExecutionsDao import org.apache.texera.dao.jooq.generated.tables.pojos.{WorkflowExecutions, User => UserPojo} import org.apache.texera.web.model.http.request.result.ResultExportRequest @@ -794,6 +795,65 @@ class WorkflowExecutionsResource { .entity(Map("error" -> "No sufficient access privilege.").asJava) .build() + /** 403 rather than the 401 above, for a caller who *can* read the workflow but may not take + * this particular data out of it. The distinction is also operational: the frontend's + * UnauthorizedHttpInterceptor reads any 401 on an authenticated request as an expired + * session and logs the user out, which is the wrong outcome for a live session that simply + * asked for something it is not entitled to. + */ + private def exportDeniedResponse(message: String): Response = + Response + .status(Response.Status.FORBIDDEN) + .`type`(MediaType.APPLICATION_JSON) + .entity(Map("error" -> message).asJava) + .build() + + /** Authorizes one result-export request, for both the local and the dataset endpoint. + * + * Three things have to hold before results leave the system: + * + * 1. the caller can read the workflow the results belong to; + * 2. the caller can reach the computing unit that produced them — `getLatestExecutionID` + * selects on `(wid, cuid)` alone, so the unit is as much a caller-chosen key as the + * workflow id is; + * 3. no requested operator carries data from a dataset whose owner marked it + * non-downloadable. `getWorkflowResultDownloadability` hands the same map to the UI, + * which greys those operators out — but that is advice to a cooperating client, and + * both export endpoints are reachable directly. + * + * @return the response to send back, or None when the export may proceed + */ + private def validateUserCanExportResult( + user: UserPojo, + request: ResultExportRequest + ): Option[Response] = { + if (!WorkflowAccessResource.hasReadAccess(request.workflowId, user.getUid)) { + Some(workflowAccessDeniedResponse) + } else if ( + ComputingUnitAccess + .getComputingUnitAccess(request.computingUnitId, user.getUid) + .eq(PrivilegeEnum.NONE) + ) { + Some(exportDeniedResponse("No sufficient access privilege to the computing unit.")) + } else { + val restrictions = getNonDownloadableOperatorMap(request.workflowId, user) + val blockingDatasets = + request.operators.flatMap(op => restrictions.getOrElse(op.id, Set.empty)).distinct + if (blockingDatasets.isEmpty) { + None + } else { + val labels = blockingDatasets.map { + case (ownerEmail, datasetName) => s"$datasetName ($ownerEmail)" + } + Some( + exportDeniedResponse( + s"Export is blocked by non-downloadable dataset(s): ${labels.mkString(", ")}" + ) + ) + } + } + } + /** Delete a group of executions */ @PUT @Consumes(Array(MediaType.APPLICATION_JSON)) @@ -857,22 +917,22 @@ class WorkflowExecutionsResource { @Path("/result/export/dataset") @RolesAllowed(Array("REGULAR", "ADMIN")) def exportResultToDataset(request: ResultExportRequest, @Auth user: SessionUser): Response = { - if (!WorkflowAccessResource.hasReadAccess(request.workflowId, user.getUser.getUid)) { - workflowAccessDeniedResponse - } else { - try { - val resultExportService = - new ResultExportService(WorkflowIdentity(request.workflowId), request.computingUnitId) - resultExportService.exportToDataset(user.user, request) - - } catch { - case ex: Exception => - Response - .status(Response.Status.INTERNAL_SERVER_ERROR) - .`type`(MediaType.APPLICATION_JSON) - .entity(Map("error" -> ex.getMessage).asJava) - .build() - } + validateUserCanExportResult(user.getUser, request) match { + case Some(denial) => denial + case None => + try { + val resultExportService = + new ResultExportService(WorkflowIdentity(request.workflowId), request.computingUnitId) + resultExportService.exportToDataset(user.user, request) + + } catch { + case ex: Exception => + Response + .status(Response.Status.INTERNAL_SERVER_ERROR) + .`type`(MediaType.APPLICATION_JSON) + .entity(Map("error" -> ex.getMessage).asJava) + .build() + } } } @@ -897,12 +957,12 @@ class WorkflowExecutionsResource { } val request = Json.parse(requestJson).as[ResultExportRequest] - if (!WorkflowAccessResource.hasReadAccess(request.workflowId, user.getUser.getUid)) { - workflowAccessDeniedResponse - } else { - val resultExportService = - new ResultExportService(WorkflowIdentity(request.workflowId), request.computingUnitId) - resultExportService.exportToLocal(request) + validateUserCanExportResult(user.getUser, request) match { + case Some(denial) => denial + case None => + val resultExportService = + new ResultExportService(WorkflowIdentity(request.workflowId), request.computingUnitId) + resultExportService.exportToLocal(request) } } catch { diff --git a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResourceSpec.scala index 7cc549004d6..aef91709e6a 100644 --- a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResourceSpec.scala +++ b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResourceSpec.scala @@ -211,12 +211,14 @@ class WorkflowExecutionsResourceSpec getDSLContext .deleteFrom(WORKFLOW_COMPUTING_UNIT) - .where(WORKFLOW_COMPUTING_UNIT.UID.eq(testUserId)) + .where(WORKFLOW_COMPUTING_UNIT.UID.between(testUserId, testUserId + 10)) .execute() + // The range, not just testUserId: cases that seed a dataset owner or a computing-unit + // owner alongside the test user would otherwise leave those rows behind for the next one. getDSLContext .deleteFrom(USER) - .where(USER.UID.eq(testUserId)) + .where(USER.UID.between(testUserId, testUserId + 10)) .execute() } @@ -226,9 +228,9 @@ class WorkflowExecutionsResourceSpec // ─── helpers ────────────────────────────────────────────────────────────── - private def insertComputingUnit(): WorkflowComputingUnit = { + private def insertComputingUnit(ownerUid: Integer = null): WorkflowComputingUnit = { val unit = new WorkflowComputingUnit - unit.setUid(testUser.getUid) + unit.setUid(if (ownerUid == null) testUser.getUid else ownerUid) unit.setName("test-unit-" + UUID.randomUUID().toString.substring(0, 8)) unit.setCreationTime(new Timestamp(System.currentTimeMillis())) unit.setType(WorkflowComputingUnitTypeEnum.local) @@ -1065,6 +1067,56 @@ class WorkflowExecutionsResourceSpec .set(WORKFLOW_USER_ACCESS.PRIVILEGE, PrivilegeEnum.READ) .execute() + /** Seeds an extra user. Uids stay within `testUserId + 10` so `cleanupTestData` reclaims + * them, which keeps a case free to reuse a uid a previous case seeded. + */ + private def insertUser(uid: Int, email: String): User = { + val u = new User + u.setUid(uid) + u.setName(s"user-$uid") + u.setEmail(email) + userDao.insert(u) + u + } + + /** Points the test workflow's `scanA` at a non-downloadable dataset owned by `ownerEmail`, + * with `downstreamB` fed from it so the BFS propagation is observable too. + */ + private def seedNonDownloadableWorkflow(ownerUid: Int, ownerEmail: String): Unit = { + insertUser(ownerUid, ownerEmail) + + val dataset = new Dataset + dataset.setOwnerUid(ownerUid) + dataset.setName("LockedDS") + dataset.setRepositoryName("repo-locked-" + UUID.randomUUID().toString.substring(0, 8)) + dataset.setIsPublic(false) + dataset.setIsDownloadable(false) + dataset.setDescription("") + dataset.setCreationTime(new Timestamp(System.currentTimeMillis())) + datasetDao.insert(dataset) + + testWorkflow.setContent( + s"""{ + | "operators": [ + | {"operatorID": "scanA", "operatorProperties": {"fileName": "/dataset/$ownerEmail/LockedDS/v1/data.csv"}}, + | {"operatorID": "downstreamB", "operatorProperties": {}} + | ], + | "links": [ + | {"source": {"operatorID": "scanA"}, "target": {"operatorID": "downstreamB"}} + | ] + |}""".stripMargin + ) + workflowDao.update(testWorkflow) + } + + private def grantComputingUnitAccess(cuid: Integer, uid: Integer = testUserId): Unit = + getDSLContext + .insertInto(COMPUTING_UNIT_USER_ACCESS) + .set(COMPUTING_UNIT_USER_ACCESS.CUID, cuid) + .set(COMPUTING_UNIT_USER_ACCESS.UID, uid) + .set(COMPUTING_UNIT_USER_ACCESS.PRIVILEGE, PrivilegeEnum.READ) + .execute() + private def userWithoutAccess(): User = { val u = new User u.setUid(testUserId + 5000) @@ -1571,9 +1623,12 @@ class WorkflowExecutionsResourceSpec // every result download, and a one-sided test cannot see a removal. it should "let every allowed role past the role gate" in { grantReadAccess() + val unit = insertComputingUnit() Seq(UserRoleEnum.REGULAR, UserRoleEnum.ADMIN).foreach { role => val response = resource.exportResultToLocal( - Json.stringify(Json.toJson(exportRequest(List(OperatorExportInfo("op-1", "csv")), 0))), + Json.stringify( + Json.toJson(exportRequest(List(OperatorExportInfo("op-1", "csv")), unit.getCuid)) + ), tokenFor(role) ) assert( @@ -1633,10 +1688,14 @@ class WorkflowExecutionsResourceSpec it should "report a missing execution for a single-operator request as a 500 JSON error" in { // One operator takes the streaming branch instead, whose "no execution" outcome is - // reported through the JSON error body rather than as an escaping exception. + // reported through the JSON error body rather than as an escaping exception. The unit is + // real and the caller owns it, so the request clears authorization and fails on the lookup. grantReadAccess() + val unit = insertComputingUnit() val response = resource.exportResultToLocal( - Json.stringify(Json.toJson(exportRequest(List(OperatorExportInfo("op-1", "csv")), 0))), + Json.stringify( + Json.toJson(exportRequest(List(OperatorExportInfo("op-1", "csv")), unit.getCuid)) + ), tokenFor(UserRoleEnum.REGULAR) ) assert(response.getStatus == 500) @@ -1668,4 +1727,111 @@ class WorkflowExecutionsResourceSpec assert(response.getStatus == Response.Status.UNAUTHORIZED.getStatusCode) } + // ─── export authorization beyond workflow read access ───────────────────── + // Read access to the workflow is not the whole gate: the execution is looked up by + // (wid, cuid), and an operator's results can carry data out of a dataset its owner + // marked non-downloadable. Both endpoints answer those two with a 403 — not the 401 + // above, which the frontend's interceptor reads as an expired session. + + private val computingUnitDenied = "No sufficient access privilege to the computing unit." + + "export authorization" should "deny a download for a computing unit the caller cannot reach" in { + grantReadAccess() + val foreignUnit = insertComputingUnit(insertUser(testUserId + 2, "cu-owner@example.com").getUid) + insertExecution(cuid = foreignUnit.getCuid) + + val response = resource.exportResultToLocal( + Json.stringify( + Json.toJson(exportRequest(List(OperatorExportInfo("op-1", "csv")), foreignUnit.getCuid)) + ), + tokenFor(UserRoleEnum.REGULAR) + ) + assert(response.getStatus == Response.Status.FORBIDDEN.getStatusCode) + assert(errorOf(response) == computingUnitDenied) + } + + it should "deny an export to dataset for a computing unit the caller cannot reach" in { + grantReadAccess() + val foreignUnit = insertComputingUnit(insertUser(testUserId + 2, "cu-owner@example.com").getUid) + insertExecution(cuid = foreignUnit.getCuid) + + val response = resource.exportResultToDataset( + exportRequest(List(OperatorExportInfo("op-1", "csv")), foreignUnit.getCuid), + session(testUser) + ) + assert(response.getStatus == Response.Status.FORBIDDEN.getStatusCode) + assert(errorOf(response) == computingUnitDenied) + } + + // Shared access, not ownership, is the predicate: a unit someone else owns but granted + // the caller READ on has to still export, or the check would break every shared unit. + it should "allow a download for a computing unit shared with the caller" in { + grantReadAccess() + val foreignUnit = insertComputingUnit(insertUser(testUserId + 2, "cu-owner@example.com").getUid) + grantComputingUnitAccess(foreignUnit.getCuid) + insertExecution(cuid = foreignUnit.getCuid) + + val response = resource.exportResultToLocal( + Json.stringify( + Json.toJson( + exportRequest( + List(OperatorExportInfo("op-1", "csv"), OperatorExportInfo("op-2", "csv")), + foreignUnit.getCuid + ) + ) + ), + tokenFor(UserRoleEnum.REGULAR) + ) + assert(response.getStatus == 200) + } + + it should "deny a download of an operator fed by a non-downloadable dataset" in { + grantReadAccess() + val unit = insertComputingUnit() + insertExecution(cuid = unit.getCuid) + seedNonDownloadableWorkflow(testUserId + 3, "locked-owner@example.com") + + // The downstream operator, not the scan itself: the restriction propagates along the + // links, and the data reaching downstreamB came out of the locked dataset. + val response = resource.exportResultToLocal( + Json.stringify( + Json.toJson(exportRequest(List(OperatorExportInfo("downstreamB", "csv")), unit.getCuid)) + ), + tokenFor(UserRoleEnum.REGULAR) + ) + assert(response.getStatus == Response.Status.FORBIDDEN.getStatusCode) + assert(errorOf(response).contains("LockedDS (locked-owner@example.com)")) + } + + it should "deny an export to dataset of an operator fed by a non-downloadable dataset" in { + grantReadAccess() + val unit = insertComputingUnit() + insertExecution(cuid = unit.getCuid) + seedNonDownloadableWorkflow(testUserId + 3, "locked-owner@example.com") + + val response = resource.exportResultToDataset( + exportRequest(List(OperatorExportInfo("scanA", "csv")), unit.getCuid), + session(testUser) + ) + assert(response.getStatus == Response.Status.FORBIDDEN.getStatusCode) + assert(errorOf(response).contains("LockedDS (locked-owner@example.com)")) + } + + // Only the operators carrying the restricted data are blocked; one restriction in the + // workflow must not close the whole workflow down. + it should "let an unrestricted operator of a restricted workflow through" in { + grantReadAccess() + val unit = insertComputingUnit() + insertExecution(cuid = unit.getCuid) + seedNonDownloadableWorkflow(testUserId + 3, "locked-owner@example.com") + + val response = resource.exportResultToLocal( + Json.stringify( + Json.toJson(exportRequest(List(OperatorExportInfo("op-1", "csv")), unit.getCuid)) + ), + tokenFor(UserRoleEnum.REGULAR) + ) + assert(response.getStatus != Response.Status.FORBIDDEN.getStatusCode) + } + }