aglinxinyuan commented on code in PR #7678:
URL: https://github.com/apache/texera/pull/7678#discussion_r3788512751
##########
amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResourceSpec.scala:
##########
@@ -1010,16 +1147,83 @@ class WorkflowExecutionsResourceSpec
assert(defaultEntry.whId == null)
}
+ // The execution this case points at DOES have a replay log, and that log's
scheme is
+ // one the storage layer rejects. An empty list is therefore only reachable
by
+ // refusing before the lookup: drop the access check and the unauthorized
caller
+ // reaches SequentialRecordStorage and throws instead of returning nothing.
"retrieveInteractionHistory" should "return an empty list when the user
lacks read access" in {
+ val exec = insertExecution(logLocation = "mock:///replay")
val result =
resource.retrieveInteractionHistory(
testWorkflowWid,
- Integer.valueOf(1),
+ exec.getEid,
session(userWithoutAccess())
)
assert(result.isEmpty)
}
+ it should "return an empty list when the requested execution does not exist"
in {
+ grantReadAccess()
+ val result = resource.retrieveInteractionHistory(
+ testWorkflowWid,
+ Integer.valueOf(Int.MaxValue),
+ session(testUser)
+ )
+ assert(result.isEmpty)
+ }
+
+ it should "return an empty list when the execution stored no replay log" in {
+ // log_location is empty, so the replay-log storage must not be opened at
all:
+ // handing "" to SequentialRecordStorage would fail rather than yield
nothing.
+ grantReadAccess()
+ val exec = insertExecution(logLocation = "")
+ val result =
+ resource.retrieveInteractionHistory(testWorkflowWid, exec.getEid,
session(testUser))
+ assert(result.isEmpty)
+ }
+
+ it should "return an empty list when log_location is NULL" in {
+ // `log_location` is nullable with no default (sql/texera_ddl.sql), so
jOOQ can hand
+ // back null here. The null half of the guard is what keeps
`null.nonEmpty` — an NPE
+ // via augmentString — from reaching the caller; the empty-string case
above cannot
+ // observe it.
+ grantReadAccess()
+ val exec = insertExecution(logLocation = null)
+ val result =
+ resource.retrieveInteractionHistory(testWorkflowWid, exec.getEid,
session(testUser))
+ assert(result.isEmpty)
+ }
+
+ // The endpoint's only real product is the list of ECM ids read out of the
replay log,
+ // and nothing in the repo observed it — so a body that always answered
`List()` was
+ // indistinguishable from a working one. Write a real two-record log and
read it back.
+ it should "return the replay destinations recorded in the execution's log,
in order" in {
+ grantReadAccess()
+ val root =
Files.createTempDirectory("workflow-executions-resource-spec-replay-")
+ try {
+ val logUri = root.resolve("logs").toUri
+ withAmberSerde {
+ val storage = new VFSRecordStorage[ReplayLogRecord](logUri)
+ // The endpoint reads the reserved "COORDINATOR" file out of the log
folder.
+ val writer = storage.getWriter("COORDINATOR")
+ try {
+
writer.writeRecord(ReplayDestination(EmbeddedControlMessageIdentity("ecm-1")))
+
writer.writeRecord(ReplayDestination(EmbeddedControlMessageIdentity("ecm-2")))
+ writer.flush()
+ } finally {
+ writer.close()
+ }
+
+ val exec = insertExecution(logLocation = logUri.toString)
+ val result =
+ resource.retrieveInteractionHistory(testWorkflowWid, exec.getEid,
session(testUser))
+ assert(result == List("ecm-1", "ecm-2"))
Review Comment:
Addressed in 13c0155. retrieveInteractionHistory now joins
WORKFLOW_EXECUTIONS through WORKFLOW_VERSION and filters by the requested wid,
so a foreign execution ID returns an empty result without opening its replay
log. I added a regression test using a foreign execution with an invalid replay
URI to pin that behavior.
##########
amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResourceSpec.scala:
##########
@@ -1080,4 +1284,350 @@ class WorkflowExecutionsResourceSpec
)
}
+ it should "throw when the stored runtime-stats URI is the empty string" in {
+ // The column is non-null here but blank, which `new URI("")` would turn
into a
+ // scheme-less URI the storage layer cannot resolve. The empty half of the
guard is
+ // the only thing that turns that into the same "no statistics" error as a
NULL.
+ grantReadAccess()
+ val exec = insertExecution(runtimeStatsUri = "")
+ assertThrows[java.util.NoSuchElementException](
+ resource.retrieveWorkflowRuntimeStatistics(testWorkflowWid, exec.getEid,
session(testUser))
+ )
+ }
+
+ it should "reject a user with no read access on the workflow" in {
+ // The URI is load-bearing: without the access check the call runs on to
the storage
+ // layer and fails there with an IllegalArgumentException over the scheme,
so only a
+ // WebApplicationException proves the request was refused up front.
+ val exec = insertExecution(runtimeStatsUri = "mock:///stats")
+ assertThrows[WebApplicationException](
+ resource.retrieveWorkflowRuntimeStatistics(
+ testWorkflowWid,
+ exec.getEid,
+ session(userWithoutAccess())
+ )
+ )
+ }
+
+ // `updateRuntimeStatsUri` already refuses to write across workflows; the
read path
+ // needs the same, or read access on any workflow would expose every other
workflow's
+ // operator ids, tuple counts and timings by execution id alone.
+ it should "refuse an execution id that belongs to a different workflow" in {
+ grantReadAccess()
+ val foreign = insertForeignExecution(runtimeStatsUri = "mock:///stats")
+ assertThrows[java.util.NoSuchElementException](
+ resource.retrieveWorkflowRuntimeStatistics(
+ testWorkflowWid,
+ foreign.getEid,
+ session(testUser)
+ )
+ )
+ }
+
+ // Per-user warehouses are off in this deployment (#6930): a read of
statistics
+ // that live in one must fail loudly and name the warehouse, because opening
it
+ // anyway resolves to the shared default and looks like data loss.
+ it should "refuse to read statistics stored in a per-user warehouse while
the feature is off" in {
+ grantReadAccess()
+ val exec = insertExecution()
+ val uri = VFSURIFactory.createRuntimeStatisticsURI(
+ WorkflowIdentity(testWorkflowWid.longValue()),
+ ExecutionIdentity(exec.getEid.longValue()),
+ warehouse = Some("byo")
+ )
+ exec.setRuntimeStatsUri(uri.toString)
+ workflowExecutionsDao.update(exec)
+
+ val ex = intercept[WarehouseUnavailableException](
+ resource.retrieveWorkflowRuntimeStatistics(testWorkflowWid, exec.getEid,
session(testUser))
+ )
+ // The whole message, not just "byo": the warehouse name is a substring of
the URI
+ // the fixture supplied, so every other refusal — including the guard's
+ // "unresolvable warehouse URI" branch — would also contain it.
+ assert(
+ ex.getMessage ==
+ "this result is stored in warehouse 'byo'; " +
+ "per-user warehouses are disabled in this deployment"
+ )
+ }
+
+ it should "report an unreadable stats URI as a URI error, not as a warehouse
refusal" in {
+ // The typed WarehouseUnavailableException is a kill-switch signal that
callers
+ // deliberately let through their catch-alls, so it must be reserved for
URIs the
+ // guard actually classifies as warehouse-scoped. Refusing every URI at
this call
+ // site would report corrupt data as "per-user warehouses are disabled".
+ grantReadAccess()
+ val exec = insertExecution(runtimeStatsUri = "mock:///stats")
+ val ex = intercept[IllegalArgumentException](
+ resource.retrieveWorkflowRuntimeStatistics(testWorkflowWid, exec.getEid,
session(testUser))
+ )
+ // WarehouseUnavailableException is an IllegalStateException, so
intercepting
+ // IllegalArgumentException already excludes it.
+ assert(ex.getMessage.contains("Invalid URI scheme"))
+ }
+
+ // ─── new: getWorkflowResultDownloadability ────────────────────────────────
+
+ private val foreignOwnerEmail = "[email protected]"
+
+ private def seedForeignDatasetOwner(): Integer = {
+ val ownerUid = Integer.valueOf(testUserId + 2)
+ getDSLContext.deleteFrom(USER).where(USER.UID.eq(ownerUid)).execute()
+ val owner = new User
+ owner.setUid(ownerUid)
+ owner.setName("restricted_ds_owner")
+ owner.setEmail(foreignOwnerEmail)
+ userDao.insert(owner)
+ ownerUid
+ }
+
+ private def seedForeignDataset(ownerUid: Integer, name: String,
downloadable: Boolean): Unit = {
+ val dataset = new Dataset
+ dataset.setOwnerUid(ownerUid)
+ dataset.setName(name)
+ dataset.setRepositoryName(s"repo-$name")
+ dataset.setIsPublic(false)
+ dataset.setIsDownloadable(downloadable)
+ dataset.setDescription("")
+ dataset.setCreationTime(new Timestamp(System.currentTimeMillis()))
+ datasetDao.insert(dataset)
+ }
+
+ private def scanOperator(operatorId: String, datasetName: String): String =
+ s"""{"operatorID": "$operatorId", "operatorProperties": """ +
+ s"""{"fileName":
"/datasets/$foreignOwnerEmail/$datasetName/v1/data.csv"}}"""
+
+ // Seeds three datasets owned by somebody other than testUser and wires the
workflow as
+ //
+ // scanA -> LockedDS (foreign, NOT downloadable) --\
+ // >-- downstreamB
+ // scanC -> LockedDS2 (foreign, NOT downloadable) --/
+ // scanD -> OpenDS (foreign, downloadable) (unrestricted, no
links)
+ //
+ // scanD is what makes the is_downloadable predicate observable — every
other dataset
+ // fixture in this file is non-downloadable, so without it "restricted" and
"foreign"
+ // are the same set. The two restricted scans meeting at downstreamB are
what make the
+ // per-operator union observable, since one restricted source cannot tell a
merge from
+ // an overwrite.
+ private def seedRestrictedWorkflow(): Unit = {
+ val ownerUid = seedForeignDatasetOwner()
+ seedForeignDataset(ownerUid, "LockedDS", downloadable = false)
+ seedForeignDataset(ownerUid, "LockedDS2", downloadable = false)
+ seedForeignDataset(ownerUid, "OpenDS", downloadable = true)
+
+ testWorkflow.setContent(
+ s"""{
+ | "operators": [
+ | ${scanOperator("scanA", "LockedDS")},
+ | ${scanOperator("scanC", "LockedDS2")},
+ | ${scanOperator("scanD", "OpenDS")},
+ | {"operatorID": "downstreamB", "operatorProperties": {}}
+ | ],
+ | "links": [
+ | {"source": {"operatorID": "scanA"}, "target": {"operatorID":
"downstreamB"}},
+ | {"source": {"operatorID": "scanC"}, "target": {"operatorID":
"downstreamB"}}
+ | ]
+ |}""".stripMargin
+ )
+ workflowDao.update(testWorkflow)
+ }
+
+ "getWorkflowResultDownloadability" should "reject a user without read
access" in {
+ assertThrows[WebApplicationException](
+ resource.getWorkflowResultDownloadability(testWorkflowWid,
session(userWithoutAccess()))
+ )
+ }
+
+ // The label format is a contract with the frontend, which renders the
strings
+ // verbatim, so it is pinned here rather than left to the caller to
reconstruct.
+ it should "label every restricted operator with 'datasetName (ownerEmail)'"
in {
+ grantReadAccess()
+ seedRestrictedWorkflow()
+
+ val response = resource.getWorkflowResultDownloadability(testWorkflowWid,
session(testUser))
+ assert(response.getStatus == 200)
+
+ val body = response.getEntity.asInstanceOf[java.util.Map[String,
Array[String]]]
+ assert(body.get("scanA").toSeq == Seq(s"LockedDS ($foreignOwnerEmail)"))
+ assert(body.get("scanC").toSeq == Seq(s"LockedDS2 ($foreignOwnerEmail)"))
+ // Two restricted scans feed downstreamB, so its entry is the union of
both; an
+ // implementation that overwrote instead of merging would list whichever
arrived
+ // last, which is why the value type is a Set.
+ assert(
+ body.get("downstreamB").toSet ==
+ Set(s"LockedDS ($foreignOwnerEmail)", s"LockedDS2
($foreignOwnerEmail)")
+ )
+ // OpenDS is foreign too, but downloadable — so scanD is not restricted at
all.
+ assert(!body.containsKey("scanD"))
+ }
+
+ // Loop workflows put a LoopEnd -> LoopStart back edge into exactly the
`links` array
+ // this endpoint walks, so a cycle is not hypothetical. Propagation stops
once an
+ // operator's restriction set stops growing; without that check the queue
cycles
+ // forever and the request thread wedges, so the call is made off-thread and
the case
+ // fails on timeout instead of hanging the suite.
+ it should "terminate on a workflow whose links form a cycle" in {
+ grantReadAccess()
+ val ownerUid = seedForeignDatasetOwner()
+ seedForeignDataset(ownerUid, "CycleDS", downloadable = false)
+ testWorkflow.setContent(
+ s"""{
+ | "operators": [
+ | ${scanOperator("scanA", "CycleDS")},
+ | {"operatorID": "b", "operatorProperties": {}},
+ | {"operatorID": "c", "operatorProperties": {}}
+ | ],
+ | "links": [
+ | {"source": {"operatorID": "scanA"}, "target": {"operatorID":
"b"}},
+ | {"source": {"operatorID": "b"}, "target": {"operatorID": "c"}},
+ | {"source": {"operatorID": "c"}, "target": {"operatorID":
"scanA"}}
+ | ]
+ |}""".stripMargin
+ )
+ workflowDao.update(testWorkflow)
+
+ val call = Future(
+ resource.getWorkflowResultDownloadability(testWorkflowWid,
session(testUser))
+ )(ExecutionContext.global)
+ val response = Await.result(call, 30.seconds)
+
+ val body = response.getEntity.asInstanceOf[java.util.Map[String,
Array[String]]]
+ assert(body.size() == 3)
+ assert(body.containsKey("scanA") && body.containsKey("b") &&
body.containsKey("c"))
+ }
+
+ // ─── new: result-export endpoints ─────────────────────────────────────────
+
+ private def exportRequest(
+ operators: List[OperatorExportInfo],
+ computingUnitId: Integer
+ ): ResultExportRequest =
+ ResultExportRequest(
+ exportType = "csv",
+ workflowId = testWorkflowWid,
+ workflowName = "export-spec-workflow",
+ operators = operators,
+ datasetIds = List.empty,
+ rowIndex = 0,
+ columnIndex = 0,
+ filename = "",
+ computingUnitId = computingUnitId.intValue()
+ )
+
+ // Mirrors what JwtAuth.jwtClaims writes at issue time, so the token below is
+ // one the production consumer accepts.
+ private def tokenFor(role: UserRoleEnum): String = {
+ val claims = new JwtClaims
+ claims.setSubject(testUser.getName)
+ claims.setClaim("userId", testUser.getUid)
+ claims.setClaim("email", testUser.getEmail)
+ claims.setClaim("role", role.name)
+ claims.setClaim("avatar", testUser.getAvatar)
+ claims.setExpirationTimeMinutesInTheFuture(10f)
+ JwtAuth.jwtToken(claims)
+ }
+
+ private def errorOf(response: Response): String =
+ response.getEntity.asInstanceOf[java.util.Map[String, String]].get("error")
+
+ // This endpoint is reached by a browser form submit, which cannot carry an
+ // Authorization header, so the JWT arrives as a form field and every failure
+ // has to come back as a JSON body rather than as an escaping exception.
+ "exportResultToLocal" should "answer an unverifiable token with a 500 JSON
error" in {
+ val response =
+
resource.exportResultToLocal(Json.stringify(Json.toJson(exportRequest(Nil,
0))), "not-a-jwt")
+ assert(response.getStatus == 500)
+ assert(errorOf(response) == "Invalid or expired token")
+ }
+
+ it should "answer a verified token whose role is below REGULAR with a 500
JSON error" in {
+ Seq(UserRoleEnum.RESTRICTED, UserRoleEnum.INACTIVE).foreach { role =>
+ val response = resource.exportResultToLocal(
+ Json.stringify(Json.toJson(exportRequest(Nil, 0))),
+ tokenFor(role)
+ )
+ assert(response.getStatus == 500, s"role $role")
+ assert(
+ errorOf(response) == "User role is not allowed to perform this
download",
+ s"role $role"
+ )
+ }
+ }
+
+ // Both allowed roles, not just REGULAR: shrinking the allow-list locks
admins out of
+ // every result download, and a one-sided test cannot see a removal.
+ it should "let every allowed role past the role gate" in {
+ grantReadAccess()
+ Seq(UserRoleEnum.REGULAR, UserRoleEnum.ADMIN).foreach { role =>
+ val response = resource.exportResultToLocal(
+
Json.stringify(Json.toJson(exportRequest(List(OperatorExportInfo("op-1",
"csv")), 0))),
+ tokenFor(role)
+ )
+ assert(
+ errorOf(response) != "User role is not allowed to perform this
download",
+ s"role $role was rejected by the role gate"
+ )
+ }
+ }
+
+ // The read grant is not required by today's code — this endpoint gates on
role only —
+ // but the request has to be legitimate on its own terms, or this case would
go red the
+ // day the missing workflow-access check is added and would block that fix.
+ it should "parse the form-encoded request and run the export once the token
checks out" in {
+ grantReadAccess()
+ val unit = insertComputingUnit()
+ insertExecution(cuid = unit.getCuid)
+
+ // Two operators, so the request takes the zip branch, which only reaches
a 200 once
+ // the workflow id and computing unit id off the parsed body find a real
execution.
+ // With either id perturbed the lookup comes back empty and the response
is a 500.
+ val response = resource.exportResultToLocal(
+ Json.stringify(
+ Json.toJson(
+ exportRequest(
+ List(OperatorExportInfo("op-1", "csv"), OperatorExportInfo("op-2",
"csv")),
+ unit.getCuid
+ )
+ )
+ ),
+ tokenFor(UserRoleEnum.REGULAR)
+ )
+ assert(response.getStatus == 200)
+ val disposition = response.getHeaderString("Content-Disposition")
+ // The name is built from the request's own workflowName, so a body the
endpoint
+ // ignored in favour of a hard-coded request would not produce it.
+ assert(disposition.startsWith("attachment;
filename=\"export-spec-workflow-"))
+ assert(disposition.endsWith(".zip\""))
+ }
+
+ 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.
+ grantReadAccess()
+ val response = resource.exportResultToLocal(
+ Json.stringify(Json.toJson(exportRequest(List(OperatorExportInfo("op-1",
"csv")), 0))),
+ tokenFor(UserRoleEnum.REGULAR)
+ )
+ assert(response.getStatus == 500)
+ assert(errorOf(response) == "Failed to export operator")
+ }
+
+ "exportResultToDataset" should "report a per-operator failure inside a 200
response" in {
+ grantReadAccess()
+ val unit = insertComputingUnit()
+ insertExecution(cuid = unit.getCuid)
Review Comment:
Addressed in 13c0155. exportResultToDataset now checks read access to
request.workflowId before constructing ResultExportService and returns 401 when
it is denied. I added a no-grant regression test that previously returned a
spurious 200 response.
##########
amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResourceSpec.scala:
##########
@@ -1080,4 +1284,350 @@ class WorkflowExecutionsResourceSpec
)
}
+ it should "throw when the stored runtime-stats URI is the empty string" in {
+ // The column is non-null here but blank, which `new URI("")` would turn
into a
+ // scheme-less URI the storage layer cannot resolve. The empty half of the
guard is
+ // the only thing that turns that into the same "no statistics" error as a
NULL.
+ grantReadAccess()
+ val exec = insertExecution(runtimeStatsUri = "")
+ assertThrows[java.util.NoSuchElementException](
+ resource.retrieveWorkflowRuntimeStatistics(testWorkflowWid, exec.getEid,
session(testUser))
+ )
+ }
+
+ it should "reject a user with no read access on the workflow" in {
+ // The URI is load-bearing: without the access check the call runs on to
the storage
+ // layer and fails there with an IllegalArgumentException over the scheme,
so only a
+ // WebApplicationException proves the request was refused up front.
+ val exec = insertExecution(runtimeStatsUri = "mock:///stats")
+ assertThrows[WebApplicationException](
+ resource.retrieveWorkflowRuntimeStatistics(
+ testWorkflowWid,
+ exec.getEid,
+ session(userWithoutAccess())
+ )
+ )
+ }
+
+ // `updateRuntimeStatsUri` already refuses to write across workflows; the
read path
+ // needs the same, or read access on any workflow would expose every other
workflow's
+ // operator ids, tuple counts and timings by execution id alone.
+ it should "refuse an execution id that belongs to a different workflow" in {
+ grantReadAccess()
+ val foreign = insertForeignExecution(runtimeStatsUri = "mock:///stats")
+ assertThrows[java.util.NoSuchElementException](
+ resource.retrieveWorkflowRuntimeStatistics(
+ testWorkflowWid,
+ foreign.getEid,
+ session(testUser)
+ )
+ )
+ }
+
+ // Per-user warehouses are off in this deployment (#6930): a read of
statistics
+ // that live in one must fail loudly and name the warehouse, because opening
it
+ // anyway resolves to the shared default and looks like data loss.
+ it should "refuse to read statistics stored in a per-user warehouse while
the feature is off" in {
+ grantReadAccess()
+ val exec = insertExecution()
+ val uri = VFSURIFactory.createRuntimeStatisticsURI(
+ WorkflowIdentity(testWorkflowWid.longValue()),
+ ExecutionIdentity(exec.getEid.longValue()),
+ warehouse = Some("byo")
+ )
+ exec.setRuntimeStatsUri(uri.toString)
+ workflowExecutionsDao.update(exec)
+
+ val ex = intercept[WarehouseUnavailableException](
+ resource.retrieveWorkflowRuntimeStatistics(testWorkflowWid, exec.getEid,
session(testUser))
+ )
+ // The whole message, not just "byo": the warehouse name is a substring of
the URI
+ // the fixture supplied, so every other refusal — including the guard's
+ // "unresolvable warehouse URI" branch — would also contain it.
+ assert(
+ ex.getMessage ==
+ "this result is stored in warehouse 'byo'; " +
+ "per-user warehouses are disabled in this deployment"
+ )
+ }
+
+ it should "report an unreadable stats URI as a URI error, not as a warehouse
refusal" in {
+ // The typed WarehouseUnavailableException is a kill-switch signal that
callers
+ // deliberately let through their catch-alls, so it must be reserved for
URIs the
+ // guard actually classifies as warehouse-scoped. Refusing every URI at
this call
+ // site would report corrupt data as "per-user warehouses are disabled".
+ grantReadAccess()
+ val exec = insertExecution(runtimeStatsUri = "mock:///stats")
+ val ex = intercept[IllegalArgumentException](
+ resource.retrieveWorkflowRuntimeStatistics(testWorkflowWid, exec.getEid,
session(testUser))
+ )
+ // WarehouseUnavailableException is an IllegalStateException, so
intercepting
+ // IllegalArgumentException already excludes it.
+ assert(ex.getMessage.contains("Invalid URI scheme"))
+ }
+
+ // ─── new: getWorkflowResultDownloadability ────────────────────────────────
+
+ private val foreignOwnerEmail = "[email protected]"
+
+ private def seedForeignDatasetOwner(): Integer = {
+ val ownerUid = Integer.valueOf(testUserId + 2)
+ getDSLContext.deleteFrom(USER).where(USER.UID.eq(ownerUid)).execute()
+ val owner = new User
+ owner.setUid(ownerUid)
+ owner.setName("restricted_ds_owner")
+ owner.setEmail(foreignOwnerEmail)
+ userDao.insert(owner)
+ ownerUid
+ }
+
+ private def seedForeignDataset(ownerUid: Integer, name: String,
downloadable: Boolean): Unit = {
+ val dataset = new Dataset
+ dataset.setOwnerUid(ownerUid)
+ dataset.setName(name)
+ dataset.setRepositoryName(s"repo-$name")
+ dataset.setIsPublic(false)
+ dataset.setIsDownloadable(downloadable)
+ dataset.setDescription("")
+ dataset.setCreationTime(new Timestamp(System.currentTimeMillis()))
+ datasetDao.insert(dataset)
+ }
+
+ private def scanOperator(operatorId: String, datasetName: String): String =
+ s"""{"operatorID": "$operatorId", "operatorProperties": """ +
+ s"""{"fileName":
"/datasets/$foreignOwnerEmail/$datasetName/v1/data.csv"}}"""
+
+ // Seeds three datasets owned by somebody other than testUser and wires the
workflow as
+ //
+ // scanA -> LockedDS (foreign, NOT downloadable) --\
+ // >-- downstreamB
+ // scanC -> LockedDS2 (foreign, NOT downloadable) --/
+ // scanD -> OpenDS (foreign, downloadable) (unrestricted, no
links)
+ //
+ // scanD is what makes the is_downloadable predicate observable — every
other dataset
+ // fixture in this file is non-downloadable, so without it "restricted" and
"foreign"
+ // are the same set. The two restricted scans meeting at downstreamB are
what make the
+ // per-operator union observable, since one restricted source cannot tell a
merge from
+ // an overwrite.
+ private def seedRestrictedWorkflow(): Unit = {
+ val ownerUid = seedForeignDatasetOwner()
+ seedForeignDataset(ownerUid, "LockedDS", downloadable = false)
+ seedForeignDataset(ownerUid, "LockedDS2", downloadable = false)
+ seedForeignDataset(ownerUid, "OpenDS", downloadable = true)
+
+ testWorkflow.setContent(
+ s"""{
+ | "operators": [
+ | ${scanOperator("scanA", "LockedDS")},
+ | ${scanOperator("scanC", "LockedDS2")},
+ | ${scanOperator("scanD", "OpenDS")},
+ | {"operatorID": "downstreamB", "operatorProperties": {}}
+ | ],
+ | "links": [
+ | {"source": {"operatorID": "scanA"}, "target": {"operatorID":
"downstreamB"}},
+ | {"source": {"operatorID": "scanC"}, "target": {"operatorID":
"downstreamB"}}
+ | ]
+ |}""".stripMargin
+ )
+ workflowDao.update(testWorkflow)
+ }
+
+ "getWorkflowResultDownloadability" should "reject a user without read
access" in {
+ assertThrows[WebApplicationException](
+ resource.getWorkflowResultDownloadability(testWorkflowWid,
session(userWithoutAccess()))
+ )
+ }
+
+ // The label format is a contract with the frontend, which renders the
strings
+ // verbatim, so it is pinned here rather than left to the caller to
reconstruct.
+ it should "label every restricted operator with 'datasetName (ownerEmail)'"
in {
+ grantReadAccess()
+ seedRestrictedWorkflow()
+
+ val response = resource.getWorkflowResultDownloadability(testWorkflowWid,
session(testUser))
+ assert(response.getStatus == 200)
+
+ val body = response.getEntity.asInstanceOf[java.util.Map[String,
Array[String]]]
+ assert(body.get("scanA").toSeq == Seq(s"LockedDS ($foreignOwnerEmail)"))
+ assert(body.get("scanC").toSeq == Seq(s"LockedDS2 ($foreignOwnerEmail)"))
+ // Two restricted scans feed downstreamB, so its entry is the union of
both; an
+ // implementation that overwrote instead of merging would list whichever
arrived
+ // last, which is why the value type is a Set.
+ assert(
+ body.get("downstreamB").toSet ==
+ Set(s"LockedDS ($foreignOwnerEmail)", s"LockedDS2
($foreignOwnerEmail)")
+ )
+ // OpenDS is foreign too, but downloadable — so scanD is not restricted at
all.
+ assert(!body.containsKey("scanD"))
+ }
+
+ // Loop workflows put a LoopEnd -> LoopStart back edge into exactly the
`links` array
+ // this endpoint walks, so a cycle is not hypothetical. Propagation stops
once an
+ // operator's restriction set stops growing; without that check the queue
cycles
+ // forever and the request thread wedges, so the call is made off-thread and
the case
+ // fails on timeout instead of hanging the suite.
+ it should "terminate on a workflow whose links form a cycle" in {
+ grantReadAccess()
+ val ownerUid = seedForeignDatasetOwner()
+ seedForeignDataset(ownerUid, "CycleDS", downloadable = false)
+ testWorkflow.setContent(
+ s"""{
+ | "operators": [
+ | ${scanOperator("scanA", "CycleDS")},
+ | {"operatorID": "b", "operatorProperties": {}},
+ | {"operatorID": "c", "operatorProperties": {}}
+ | ],
+ | "links": [
+ | {"source": {"operatorID": "scanA"}, "target": {"operatorID":
"b"}},
+ | {"source": {"operatorID": "b"}, "target": {"operatorID": "c"}},
+ | {"source": {"operatorID": "c"}, "target": {"operatorID":
"scanA"}}
+ | ]
+ |}""".stripMargin
+ )
+ workflowDao.update(testWorkflow)
+
+ val call = Future(
+ resource.getWorkflowResultDownloadability(testWorkflowWid,
session(testUser))
+ )(ExecutionContext.global)
+ val response = Await.result(call, 30.seconds)
+
+ val body = response.getEntity.asInstanceOf[java.util.Map[String,
Array[String]]]
+ assert(body.size() == 3)
+ assert(body.containsKey("scanA") && body.containsKey("b") &&
body.containsKey("c"))
+ }
+
+ // ─── new: result-export endpoints ─────────────────────────────────────────
+
+ private def exportRequest(
+ operators: List[OperatorExportInfo],
+ computingUnitId: Integer
+ ): ResultExportRequest =
+ ResultExportRequest(
+ exportType = "csv",
+ workflowId = testWorkflowWid,
+ workflowName = "export-spec-workflow",
+ operators = operators,
+ datasetIds = List.empty,
+ rowIndex = 0,
+ columnIndex = 0,
+ filename = "",
+ computingUnitId = computingUnitId.intValue()
+ )
+
+ // Mirrors what JwtAuth.jwtClaims writes at issue time, so the token below is
+ // one the production consumer accepts.
+ private def tokenFor(role: UserRoleEnum): String = {
+ val claims = new JwtClaims
+ claims.setSubject(testUser.getName)
+ claims.setClaim("userId", testUser.getUid)
+ claims.setClaim("email", testUser.getEmail)
+ claims.setClaim("role", role.name)
+ claims.setClaim("avatar", testUser.getAvatar)
+ claims.setExpirationTimeMinutesInTheFuture(10f)
+ JwtAuth.jwtToken(claims)
+ }
+
+ private def errorOf(response: Response): String =
+ response.getEntity.asInstanceOf[java.util.Map[String, String]].get("error")
+
+ // This endpoint is reached by a browser form submit, which cannot carry an
+ // Authorization header, so the JWT arrives as a form field and every failure
+ // has to come back as a JSON body rather than as an escaping exception.
+ "exportResultToLocal" should "answer an unverifiable token with a 500 JSON
error" in {
+ val response =
+
resource.exportResultToLocal(Json.stringify(Json.toJson(exportRequest(Nil,
0))), "not-a-jwt")
+ assert(response.getStatus == 500)
+ assert(errorOf(response) == "Invalid or expired token")
+ }
+
+ it should "answer a verified token whose role is below REGULAR with a 500
JSON error" in {
+ Seq(UserRoleEnum.RESTRICTED, UserRoleEnum.INACTIVE).foreach { role =>
+ val response = resource.exportResultToLocal(
+ Json.stringify(Json.toJson(exportRequest(Nil, 0))),
+ tokenFor(role)
+ )
+ assert(response.getStatus == 500, s"role $role")
+ assert(
+ errorOf(response) == "User role is not allowed to perform this
download",
+ s"role $role"
+ )
+ }
+ }
+
+ // Both allowed roles, not just REGULAR: shrinking the allow-list locks
admins out of
+ // every result download, and a one-sided test cannot see a removal.
+ it should "let every allowed role past the role gate" in {
+ grantReadAccess()
+ Seq(UserRoleEnum.REGULAR, UserRoleEnum.ADMIN).foreach { role =>
+ val response = resource.exportResultToLocal(
+
Json.stringify(Json.toJson(exportRequest(List(OperatorExportInfo("op-1",
"csv")), 0))),
+ tokenFor(role)
+ )
+ assert(
+ errorOf(response) != "User role is not allowed to perform this
download",
+ s"role $role was rejected by the role gate"
+ )
+ }
+ }
+
+ // The read grant is not required by today's code — this endpoint gates on
role only —
+ // but the request has to be legitimate on its own terms, or this case would
go red the
+ // day the missing workflow-access check is added and would block that fix.
Review Comment:
Addressed in 13c0155. After validating the JWT and role, exportResultToLocal
now verifies workflow read access before constructing ResultExportService and
returns 401 when it is denied. I added a seeded-execution, two-operator
regression test with a valid REGULAR token but no workflow grant.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
