Yicong-Huang commented on code in PR #6447:
URL: https://github.com/apache/texera/pull/6447#discussion_r3741174276
##########
amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowResource.scala:
##########
@@ -782,6 +787,41 @@ class WorkflowResource extends LazyLogging {
.fetchOneInto(classOf[String])
}
+ @POST
+ @Consumes(Array(MediaType.APPLICATION_JSON))
+ @RolesAllowed(Array("REGULAR", "ADMIN"))
+ @Path("/{wid}/export/drive")
+ def exportWorkflowToDrive(
+ @PathParam("wid") wid: Integer,
+ request: WorkflowResource.DriveExportRequest,
+ @Auth sessionUser: SessionUser
+ ): Unit = {
+ if
(!request.sessionUri.startsWith("https://www.googleapis.com/upload/drive/";))
+ throw new BadRequestException("Invalid session URI")
+ if (!WorkflowAccessResource.hasReadAccess(wid, sessionUser.getUid))
+ throw new ForbiddenException("No sufficient access privilege.")
+ val content = workflowDao.fetchOneByWid(wid).getContent.getBytes("UTF-8")
+ if (content.length > WorkflowResource.DRIVE_EXPORT_MAX_BYTES)
+ throw new BadRequestException("Workflow content exceeds the 500 MB
export limit.")
+ val conn = new
URL(request.sessionUri).openConnection().asInstanceOf[HttpURLConnection]
+ try {
+ conn.setDoOutput(true)
+ conn.setRequestMethod("PUT")
+ conn.setRequestProperty("Content-Type", "application/json")
+ conn.setRequestProperty("Content-Length", content.length.toString)
+ conn.getOutputStream.write(content)
+ val code = conn.getResponseCode
+ if (code < 200 || code >= 300) {
+ val body = Option(conn.getErrorStream).map(s => new
String(s.readAllBytes())).getOrElse("")
+ if (body.contains("storageQuotaExceeded"))
+ throw new ForbiddenException("Google Drive storage quota exceeded.")
+ throw new InternalServerErrorException(s"Google Drive upload failed:
$body")
Review Comment:
This is the third copy of the same Drive upload;
`DatasetResource.scala:1319` and `:1425` carry the other two. They have already
drifted. Both DatasetResource copies return `502 BAD_GATEWAY` with just the
status code. This one returns 500 and echoes Google's raw error body to the
caller — so our client is told we broke, not the upstream.
I'd extract one helper holding the allowlist check, the PUT and the status
handling. Patching the three copies to agree instead leaves the next edit free
to split them again.
##########
file-service/src/main/scala/org/apache/texera/service/resource/DatasetResource.scala:
##########
@@ -1298,6 +1305,187 @@ class DatasetResource extends LazyLogging {
}
}
+ @POST
+ @RolesAllowed(Array("REGULAR", "ADMIN"))
+ @Path("/{did}/drive-export")
+ @Consumes(Array(MediaType.APPLICATION_JSON))
+ def exportDatasetToDrive(
+ @PathParam("did") did: Integer,
+ @QueryParam("dvid") dvid: Integer,
+ @QueryParam("latest") latest: java.lang.Boolean,
+ request: DriveExportRequest,
+ @Auth user: SessionUser
+ ): Response = {
+ if
(!request.sessionUri.startsWith("https://www.googleapis.com/upload/drive/";)) {
+ throw new BadRequestException("Invalid session URI")
+ }
+ withTransaction(context) { ctx =>
Review Comment:
Confirming @xuang7's thread rather than reopening it. At head `febaff79`
this `withTransaction` still closes at line 1417. The LakeFS downloads
(`:1373`), the in-memory zip (`:1367`) and the Drive PUT (`:1387-1414`) all run
inside it, holding one pooled connection for the whole transfer with no read
timeout.
`getDatasetVersionZip` avoids this by returning a `StreamingOutput`
(`:1278-1301`). The full resumable-upload rework isn't needed here: keep the
access checks and `retrieveObjectsOfVersion` inside the transaction, and move
everything from `val totalBytes` onward outside it.
##########
file-service/src/main/scala/org/apache/texera/service/resource/DatasetResource.scala:
##########
@@ -1298,6 +1305,187 @@ class DatasetResource extends LazyLogging {
}
}
+ @POST
+ @RolesAllowed(Array("REGULAR", "ADMIN"))
+ @Path("/{did}/drive-export")
+ @Consumes(Array(MediaType.APPLICATION_JSON))
+ def exportDatasetToDrive(
+ @PathParam("did") did: Integer,
+ @QueryParam("dvid") dvid: Integer,
+ @QueryParam("latest") latest: java.lang.Boolean,
+ request: DriveExportRequest,
+ @Auth user: SessionUser
+ ): Response = {
+ if
(!request.sessionUri.startsWith("https://www.googleapis.com/upload/drive/";)) {
+ throw new BadRequestException("Invalid session URI")
+ }
+ withTransaction(context) { ctx =>
+ if ((dvid != null && latest != null) || (dvid == null && latest ==
null)) {
+ throw new BadRequestException("Specify exactly one: dvid=<ID> OR
latest=true")
+ }
+
+ val uid = user.getUid
+ if (!userHasReadAccess(ctx, did, uid)) {
+ throw new ForbiddenException(ERR_USER_HAS_NO_ACCESS_TO_DATASET_MESSAGE)
+ }
+
+ val dataset = getDatasetByID(ctx, did)
+ if (!userOwnDataset(ctx, did, uid) && !dataset.getIsDownloadable) {
+ throw new ForbiddenException("Dataset download is not allowed")
+ }
+
+ val datasetVersion = if (dvid != null) {
+ getDatasetVersionByID(ctx, dvid)
+ } else {
+ getLatestDatasetVersion(ctx, did).getOrElse(
+ throw new NotFoundException(ERR_DATASET_VERSION_NOT_FOUND_MESSAGE)
+ )
+ }
Review Comment:
`?latest=false` reaches this branch and exports the latest version. The
guard above only rejects "both set" and "neither set", so a non-null `false`
falls straight through — an explicit "not the latest one" is honoured as "yes,
the latest one". `getDatasetVersionZip` gates on `Boolean.TRUE.equals` and
rejects everything else (`:1253`, `:1258`).
```suggestion
} else if (java.lang.Boolean.TRUE.equals(latest)) {
getLatestDatasetVersion(ctx, did).getOrElse(
throw new NotFoundException(ERR_DATASET_VERSION_NOT_FOUND_MESSAGE)
)
} else {
throw new BadRequestException("Specify exactly one: dvid=<ID> OR
latest=true")
}
```
##########
amber/src/main/scala/org/apache/texera/web/resource/auth/GoogleAuthResource.scala:
##########
@@ -48,11 +48,17 @@ object GoogleAuthResource {
@Path("/auth/google")
class GoogleAuthResource {
final private lazy val clientId = UserSystemConfig.googleClientId
+ final private lazy val driveApiKey = UserSystemConfig.googleApiKey
@GET
@Path("/clientid")
def getClientId: String = clientId
+ @GET
+ @Path("/drive/apikey")
+ @Produces(Array(MediaType.TEXT_PLAIN))
+ def getDriveApiKey: String = driveApiKey
Review Comment:
This endpoint has no `@RolesAllowed`. Amber is also the one service that
doesn't call `RoleAnnotationEnforcer.enforce`, so nothing rejects an anonymous
request — any caller can read the deployment's Google API key and spend its
quota.
The neighbouring `getClientId` is unannotated for a reason: the OAuth client
id must be readable before login. The Picker key is only needed after login, so
it doesn't inherit that. Adding `@RolesAllowed(Array("REGULAR", "ADMIN"))` also
needs the `javax.annotation.security.RolesAllowed` import.
##########
file-service/src/test/scala/org/apache/texera/service/resource/DatasetResourceSpec.scala:
##########
@@ -2752,6 +2752,95 @@ class DatasetResourceSpec
fetchPartRows(uploadId) shouldBe empty
}
+ //
===========================================================================
+ // exportDatasetToDrive tests
+ //
===========================================================================
+
+ "exportDatasetToDrive" should "reject a session URI that does not start with
the googleapis upload prefix" in {
+ val request = DatasetResource.DriveExportRequest("http://test.com/upload";)
+ intercept[BadRequestException] {
+ datasetResource.exportDatasetToDrive(baseDataset.getDid, null, null,
request, sessionUser)
Review Comment:
This test passes whether or not the session-URI check exists. With `dvid`
and `latest` both null it trips the "Specify exactly one" guard first, which
throws the same `BadRequestException`. The call is byte-identical to the one at
`:2782` that tests that other guard, so `exportDatasetToDrive`'s SSRF control
is currently unverified.
```suggestion
val latest: java.lang.Boolean = true
datasetResource.exportDatasetToDrive(baseDataset.getDid, null, latest,
request, sessionUser)
```
##########
file-service/src/main/scala/org/apache/texera/service/resource/DatasetResource.scala:
##########
@@ -1298,6 +1305,187 @@ class DatasetResource extends LazyLogging {
}
}
+ @POST
+ @RolesAllowed(Array("REGULAR", "ADMIN"))
+ @Path("/{did}/drive-export")
+ @Consumes(Array(MediaType.APPLICATION_JSON))
+ def exportDatasetToDrive(
+ @PathParam("did") did: Integer,
+ @QueryParam("dvid") dvid: Integer,
+ @QueryParam("latest") latest: java.lang.Boolean,
+ request: DriveExportRequest,
+ @Auth user: SessionUser
+ ): Response = {
+ if
(!request.sessionUri.startsWith("https://www.googleapis.com/upload/drive/";)) {
+ throw new BadRequestException("Invalid session URI")
+ }
+ withTransaction(context) { ctx =>
+ if ((dvid != null && latest != null) || (dvid == null && latest ==
null)) {
+ throw new BadRequestException("Specify exactly one: dvid=<ID> OR
latest=true")
+ }
+
+ val uid = user.getUid
+ if (!userHasReadAccess(ctx, did, uid)) {
+ throw new ForbiddenException(ERR_USER_HAS_NO_ACCESS_TO_DATASET_MESSAGE)
+ }
+
+ val dataset = getDatasetByID(ctx, did)
+ if (!userOwnDataset(ctx, did, uid) && !dataset.getIsDownloadable) {
+ throw new ForbiddenException("Dataset download is not allowed")
+ }
+
+ val datasetVersion = if (dvid != null) {
+ getDatasetVersionByID(ctx, dvid)
+ } else {
+ getLatestDatasetVersion(ctx, did).getOrElse(
+ throw new NotFoundException(ERR_DATASET_VERSION_NOT_FOUND_MESSAGE)
+ )
+ }
+
+ val repositoryName = dataset.getRepositoryName
+ val versionHash = datasetVersion.getVersionHash
+ val objects = withLakeFSErrorHandling(
+ s"listing files of version '$versionHash' of dataset
'${dataset.getName}'"
+ ) {
+ LakeFSStorageClient.retrieveObjectsOfVersion(repositoryName,
versionHash)
+ }
+
+ if (objects.isEmpty) {
+ throw new NotFoundException(s"No objects found in version
$versionHash")
+ }
+
+ val totalBytes = objects.map(_.getSizeBytes.longValue()).sum
+ // TODO: remove this limit once chunked resumable upload to Drive is
implemented
+ if (totalBytes > DRIVE_EXPORT_MAX_DATASET_BYTES) {
+ throw new BadRequestException("Dataset version exceeds the 500 MB
export limit.")
+ }
+ if (totalBytes > DRIVE_EXPORT_MAX_FILE_BYTES) {
+ throw new ForbiddenException("Dataset version exceeds the 5 TB Google
Drive export limit.")
+ }
+
+ // Build the zip in memory
+ val baos = new ByteArrayOutputStream()
+ val zipOut = new java.util.zip.ZipOutputStream(baos)
+ try {
+ objects.foreach { obj =>
+ val filePath = obj.getPath
+ val file = withLakeFSErrorHandling(s"downloading file '$filePath'
for Drive export") {
+ LakeFSStorageClient.getFileFromRepo(repositoryName, versionHash,
filePath)
+ }
+ zipOut.putNextEntry(new java.util.zip.ZipEntry(filePath))
+ Files.copy(Paths.get(file.toURI), zipOut)
+ zipOut.closeEntry()
+ }
+ } finally {
+ zipOut.close()
+ }
+
+ val zipBytes = baos.toByteArray
+ val sessionUri = request.sessionUri
+
+ // PUT the zip to the Google Drive session URI
+ val conn = new
URL(sessionUri).openConnection().asInstanceOf[HttpURLConnection]
+ try {
+ conn.setDoOutput(true)
+ conn.setRequestMethod("PUT")
+ conn.setRequestProperty("Content-Type", "application/zip")
+ conn.setFixedLengthStreamingMode(zipBytes.length)
+ val out = conn.getOutputStream
+ out.write(zipBytes)
+ out.close()
+
+ val code = conn.getResponseCode
+ if (code < 200 || code >= 300) {
+ val body =
+ Option(conn.getErrorStream).map(s => new
String(s.readAllBytes())).getOrElse("")
+ if (body.contains("storageQuotaExceeded")) {
+ throw new ForbiddenException("Google Drive storage quota
exceeded.")
+ }
+ throw new WebApplicationException(
+ s"Google Drive upload failed with HTTP $code",
+ Response.Status.BAD_GATEWAY
+ )
+ }
+ } finally {
+ conn.disconnect()
+ }
+
+ Response.ok(Map("message" -> "Dataset exported to Google Drive")).build()
+ }
+ }
+
+ @POST
+ @RolesAllowed(Array("REGULAR", "ADMIN"))
+ @Path("/drive-export/file")
+ @Consumes(Array(MediaType.APPLICATION_JSON))
+ def exportFileToDrive(
+ request: DriveFileExportRequest,
+ @Auth user: SessionUser
+ ): Response = {
+ if
(!request.sessionUri.startsWith("https://www.googleapis.com/upload/drive/";)) {
+ throw new BadRequestException("Invalid session URI")
+ }
+ val uid = user.getUid
+ resolveDatasetAndPath(request.filePath, null, null, uid) match {
+ case Left(errorResponse) => errorResponse
+ case Right((repositoryName, commitHash, filePath)) =>
+ val presignedUrl =
+ withLakeFSErrorHandling(s"generating presigned URL for file
'$filePath'") {
+ LakeFSStorageClient.getFilePresignedUrl(repositoryName,
commitHash, filePath)
+ }
+
+ val minioConn = new
URL(presignedUrl).openConnection().asInstanceOf[HttpURLConnection]
+ try {
+ minioConn.setRequestMethod("GET")
+ val contentLength = minioConn.getContentLengthLong
+ if (contentLength > DRIVE_EXPORT_MAX_FILE_BYTES) {
+ throw new ForbiddenException("File exceeds the 5 TB Google Drive
export limit.")
+ }
+ val minioStream = minioConn.getInputStream
+
+ val driveConn =
+ new
URL(request.sessionUri).openConnection().asInstanceOf[HttpURLConnection]
+ try {
+ driveConn.setDoOutput(true)
+ driveConn.setRequestMethod("PUT")
+ driveConn.setRequestProperty("Content-Type",
"application/octet-stream")
+ if (contentLength > 0) {
+ driveConn.setFixedLengthStreamingMode(contentLength)
+ val out = driveConn.getOutputStream
+ minioStream.transferTo(out)
+ out.close()
+ } else {
+ val bytes = minioStream.readAllBytes()
Review Comment:
A question rather than a claim. This branch runs when `getContentLengthLong`
returns `-1`. By then the `DRIVE_EXPORT_MAX_FILE_BYTES` check at line 1441 has
already passed, since `-1` is not greater than the cap, so `readAllBytes()`
pulls the whole object onto the heap unbounded.
Can a MinIO presigned GET omit `Content-Length`? If it can, this path needs
its own cap; if not, the branch is dead.
##########
amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowResource.scala:
##########
@@ -782,6 +787,41 @@ class WorkflowResource extends LazyLogging {
.fetchOneInto(classOf[String])
}
+ @POST
+ @Consumes(Array(MediaType.APPLICATION_JSON))
+ @RolesAllowed(Array("REGULAR", "ADMIN"))
+ @Path("/{wid}/export/drive")
+ def exportWorkflowToDrive(
+ @PathParam("wid") wid: Integer,
+ request: WorkflowResource.DriveExportRequest,
+ @Auth sessionUser: SessionUser
+ ): Unit = {
+ if
(!request.sessionUri.startsWith("https://www.googleapis.com/upload/drive/";))
+ throw new BadRequestException("Invalid session URI")
+ if (!WorkflowAccessResource.hasReadAccess(wid, sessionUser.getUid))
+ throw new ForbiddenException("No sufficient access privilege.")
+ val content = workflowDao.fetchOneByWid(wid).getContent.getBytes("UTF-8")
+ if (content.length > WorkflowResource.DRIVE_EXPORT_MAX_BYTES)
+ throw new BadRequestException("Workflow content exceeds the 500 MB
export limit.")
+ val conn = new
URL(request.sessionUri).openConnection().asInstanceOf[HttpURLConnection]
+ try {
+ conn.setDoOutput(true)
+ conn.setRequestMethod("PUT")
+ conn.setRequestProperty("Content-Type", "application/json")
+ conn.setRequestProperty("Content-Length", content.length.toString)
+ conn.getOutputStream.write(content)
Review Comment:
`HttpURLConnection` writes its own `Content-Length` from the buffered body,
so this header never takes effect and the payload is buffered rather than
streamed. Both sibling endpoints in this PR use `setFixedLengthStreamingMode`
and close the stream (`DatasetResource.scala:1392`/`:1395`, `:1453`/`:1456`);
this one does neither.
```suggestion
conn.setFixedLengthStreamingMode(content.length)
val out = conn.getOutputStream
out.write(content)
out.close()
```
##########
common/config/src/main/resources/user-system.conf:
##########
@@ -27,6 +27,12 @@ user-sys {
clientId = ""
clientId = ${?USER_SYS_GOOGLE_CLIENT_ID}
+ clientSecret = ""
+ clientSecret = ${?USER_SYS_GOOGLE_CLIENT_SECRET}
+
+ apiKey = ""
+ apiKey = ${?USER_SYS_GOOGLE_API_KEY}
Review Comment:
`USER_SYS_GOOGLE_API_KEY` needs the same three companion entries its sibling
`USER_SYS_GOOGLE_CLIENT_ID` has: `EnvironmentalVariable.scala:122`,
`bin/k8s/values.yaml:321` and `bin/k8s/values-development.yaml:347`.
Without the k8s entries the variable is never injected. `googleApiKey` then
resolves to `""` and `/auth/google/drive/apikey` serves an empty string. So the
Picker breaks in Kubernetes only, while local dev keeps working because you
export the variable by hand.
--
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]
