This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-7760-bf0e7779ecbe64a918beadf20181306105449677 in repository https://gitbox.apache.org/repos/asf/texera.git
commit de5d7d1043b87a1d11b9db9f6dbde82302ada6ff Author: Tanishq Gandhi <[email protected]> AuthorDate: Wed Aug 19 18:45:54 2026 +0000 refactor(file-service): share the dataset access and naming rules (#7760) ### What changes were proposed in this PR? Datasets and models need the same three rules: who owns a resource, who it is shared with, and whether a name is already taken. Today those rules live inside the dataset code, so adding models means copying them. This PR moves them into one shared implementation and points datasets at it. Since jOOQ generates an unrelated class per table, there is no common parent to inherit from — instead a small descriptor says which columns to look at, and the shared code works off that. The same duplication sat one layer down: `DatasetFileDocument` and `ModelFileDocument` each existed only to supply a presign-download URL, since the read logic was already shared in `LakeFSFileDocument`. That class now takes the resource type and works out its own endpoint, so both subclasses are deleted and one document class serves every resource type. `DatasetFileDocument`'s only other method deleted files from the pre-LakeFS local git store, which nothing reaches any more. No behaviour change for datasets, apart from one fix: `GET /access/dataset/owner/{did}` and `GET /access/dataset/list/{did}` had no authorization, so any signed-in user could read any dataset's owner email and full share list. Both now require read access. Anyone who could already see the dataset is unaffected. ### Any related issues, documentation, discussions? Prepares the shared layer for #6498 (umbrella #6494). ### How was this PR tested? The existing dataset suites are the regression evidence, since this changes how the dataset code is wired without changing what it does: - `sbt "FileService/test"` — 236/236, including `DatasetResourceSpec` (118) - `sbt "WorkflowCore/test"` — 773/773 - `scalafmtCheckAll` and `scalafixAll --check` clean `DatasetAccessResourceSpec` gains tests for the two guards. One existing expectation changed: a nonexistent id returned an empty string and now returns 403. ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Claude Opus 5) --------- Co-authored-by: Claude Opus 5 <[email protected]> --- .../amber/core/storage/DocumentFactory.scala | 8 +- .../core/storage/model/DatasetFileDocument.scala | 64 ---- .../core/storage/model/LakeFSFileDocument.scala | 42 ++- .../core/storage/model/ModelFileDocument.scala | 43 --- .../amber/core/storage/DocumentFactorySpec.scala | 23 +- ...mentSpec.scala => LakeFSFileDocumentSpec.scala} | 56 ++-- .../service/resource/DatasetAccessResource.scala | 158 +++------- .../texera/service/resource/DatasetResource.scala | 135 ++------- .../texera/service/resource/ResourceAccess.scala | 321 +++++++++++++++++++++ .../texera/service/resource/ResourceNaming.scala | 99 +++++++ .../texera/service/resource/ResourceTables.scala | 69 +++++ .../resource/DatasetAccessResourceSpec.scala | 60 +++- 12 files changed, 696 insertions(+), 382 deletions(-) diff --git a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/DocumentFactory.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/DocumentFactory.scala index dfc63dc6a0..8481906c9c 100644 --- a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/DocumentFactory.scala +++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/DocumentFactory.scala @@ -60,8 +60,8 @@ object DocumentFactory { */ def openReadonlyDocument(fileUri: URI): ReadonlyVirtualDocument[_] = { fileUri.getScheme match { - case DATASET_FILE_URI_SCHEME => new DatasetFileDocument(fileUri) - case MODEL_FILE_URI_SCHEME => new ModelFileDocument(fileUri) + case DATASET_FILE_URI_SCHEME => new LakeFSFileDocument(fileUri, ResourceType.Datasets) + case MODEL_FILE_URI_SCHEME => new LakeFSFileDocument(fileUri, ResourceType.Models) case "file" => new ReadonlyLocalFileDocument(fileUri) case unsupportedScheme => throw new UnsupportedOperationException( @@ -184,8 +184,8 @@ object DocumentFactory { */ def openDocument(uri: URI): (VirtualDocument[_], Option[Schema]) = { uri.getScheme match { - case DATASET_FILE_URI_SCHEME => (new DatasetFileDocument(uri), None) - case MODEL_FILE_URI_SCHEME => (new ModelFileDocument(uri), None) + case DATASET_FILE_URI_SCHEME => (new LakeFSFileDocument(uri, ResourceType.Datasets), None) + case MODEL_FILE_URI_SCHEME => (new LakeFSFileDocument(uri, ResourceType.Models), None) case VFS_FILE_URI_SCHEME => val IcebergLocation(warehouse, namespace, storageKey) = resolveIcebergLocation(uri) diff --git a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/model/DatasetFileDocument.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/model/DatasetFileDocument.scala deleted file mode 100644 index f2862c87ee..0000000000 --- a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/model/DatasetFileDocument.scala +++ /dev/null @@ -1,64 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -package org.apache.texera.amber.core.storage.model - -import org.apache.texera.common.config.EnvironmentalVariable -import org.apache.texera.amber.core.storage.model.DatasetFileDocument.fileServiceGetPresignURLEndpoint -import org.apache.texera.amber.core.storage.util.dataset.GitVersionControlLocalFileStorage - -import java.net.URI -import java.nio.file.Path - -object DatasetFileDocument { - // The endpoint of getting presigned url from the file service, also stored in the environment vars. - lazy val fileServiceGetPresignURLEndpoint: String = - sys.env - .getOrElse( - EnvironmentalVariable.ENV_FILE_SERVICE_GET_DATASET_PRESIGNED_URL_ENDPOINT, - "http://localhost:9092/api/dataset/presign-download"; - ) - .trim -} - -private[storage] class DatasetFileDocument(uri: URI) - extends LakeFSFileDocument(uri, fileServiceGetPresignURLEndpoint) { - - override def clear(): Unit = { - // first remove the temporary file (handled by the shared base) - super.clear() - - lazy val datasetsRootPath = - Path - .of(sys.env.getOrElse("TEXERA_HOME", ".")) - .resolve("amber") - .resolve("user-resources") - .resolve("datasets") - - def getDatasetPath(did: Integer): Path = { - datasetsRootPath.resolve(did.toString) - } - - // then remove the dataset file from the local git-backed storage - GitVersionControlLocalFileStorage.removeFileFromRepo( - getDatasetPath(0), - getDatasetPath(0).resolve(fileRelativePath) - ) - } -} diff --git a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/model/LakeFSFileDocument.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/model/LakeFSFileDocument.scala index 044437889e..b821ff72f5 100644 --- a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/model/LakeFSFileDocument.scala +++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/model/LakeFSFileDocument.scala @@ -21,6 +21,7 @@ package org.apache.texera.amber.core.storage.model import com.typesafe.scalalogging.LazyLogging import org.apache.texera.common.config.EnvironmentalVariable +import org.apache.texera.amber.core.storage.ResourceType import org.apache.texera.amber.core.storage.model.LakeFSFileDocument.userJwtToken import org.apache.texera.amber.core.storage.util.LakeFSStorageClient @@ -36,21 +37,50 @@ object LakeFSFileDocument { // In the local development or other architectures, this token can be empty. lazy val userJwtToken: String = sys.env.getOrElse(EnvironmentalVariable.ENV_USER_JWT_TOKEN, "").trim + + private lazy val datasetPresignEndpoint: String = + sys.env + .getOrElse( + EnvironmentalVariable.ENV_FILE_SERVICE_GET_DATASET_PRESIGNED_URL_ENDPOINT, + "http://localhost:9092/api/dataset/presign-download"; + ) + .trim + + private lazy val modelPresignEndpoint: String = + sys.env + .getOrElse( + EnvironmentalVariable.ENV_FILE_SERVICE_GET_MODEL_PRESIGNED_URL_ENDPOINT, + "http://localhost:9092/api/model/presign-download"; + ) + .trim + + /** + * The file-service presign-download endpoint serving this resource type. Each resource type + * owns an endpoint because they enforce different access control + */ + def presignEndpointOf(resourceType: ResourceType.Value): String = + resourceType match { + case ResourceType.Datasets => datasetPresignEndpoint + case ResourceType.Models => modelPresignEndpoint + } } /** * A read-only document over a single file stored in a LakeFS repository, addressed by the URI - * {scheme}:///{repositoryName}/{versionHash}/{fileRelativePath}. This is the shared behavior - * for every versioned-file resource (datasets, models, …): the file bytes are fetched via a - * presigned URL, falling back to a direct LakeFS fetch. + * {scheme}:///{repositoryName}/{versionHash}/{fileRelativePath}. + * + * Every versioned-file resource (datasets, models, …) reads its files the same way — fetch the + * bytes through a presigned URL, falling back to a direct LakeFS fetch * - * @param uri the resolved {scheme}:///{repositoryName}/{versionHash}/{file} URI - * @param presignEndpoint the file-service presign-download endpoint for this resource kind + * @param uri the resolved {scheme}:///{repositoryName}/{versionHash}/{file} URI + * @param resourceType which resource this file belongs to, selecting the presign endpoint */ -private[storage] abstract class LakeFSFileDocument(uri: URI, presignEndpoint: String) +private[storage] class LakeFSFileDocument(uri: URI, val resourceType: ResourceType.Value) extends VirtualDocument[Nothing] with OnVersionedFileResource with LazyLogging { + + private val presignEndpoint: String = LakeFSFileDocument.presignEndpointOf(resourceType) // Utility function to parse and decode URI segments into individual components private def parseUri(uri: URI): (String, String, Path) = { val segments = Paths.get(uri.getPath).iterator().asScala.map(_.toString).toArray diff --git a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/model/ModelFileDocument.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/model/ModelFileDocument.scala deleted file mode 100644 index e67ec92992..0000000000 --- a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/model/ModelFileDocument.scala +++ /dev/null @@ -1,43 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -package org.apache.texera.amber.core.storage.model - -import org.apache.texera.common.config.EnvironmentalVariable -import org.apache.texera.amber.core.storage.model.ModelFileDocument.fileServiceGetModelPresignURLEndpoint - -import java.net.URI - -object ModelFileDocument { - // The endpoint of getting a presigned url for a model file from the file service. - lazy val fileServiceGetModelPresignURLEndpoint: String = - sys.env - .getOrElse( - EnvironmentalVariable.ENV_FILE_SERVICE_GET_MODEL_PRESIGNED_URL_ENDPOINT, - "http://localhost:9092/api/model/presign-download"; - ) - .trim -} - -/** - * A read-only document over a single file in a model's LakeFS repository (`model-{mid}`), - * addressed by a `model:///{repositoryName}/{versionHash}/{fileRelativePath}` URI. - */ -private[storage] class ModelFileDocument(uri: URI) - extends LakeFSFileDocument(uri, fileServiceGetModelPresignURLEndpoint) diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/DocumentFactorySpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/DocumentFactorySpec.scala index eb18e87762..a957819b31 100644 --- a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/DocumentFactorySpec.scala +++ b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/DocumentFactorySpec.scala @@ -20,8 +20,7 @@ package org.apache.texera.amber.core.storage import org.apache.texera.amber.core.storage.model.{ - DatasetFileDocument, - ModelFileDocument, + LakeFSFileDocument, OnVersionedFileResource, ReadonlyLocalFileDocument, VirtualDocument @@ -134,17 +133,19 @@ class DocumentFactorySpec extends AnyFlatSpec with Matchers with BeforeAndAfterA private val versionHash = "97fd4c2a755b69b7c66d322eab40b7e5c2ad5d10" - "openReadonlyDocument" should "return a DatasetFileDocument for the dataset scheme" in { + "openReadonlyDocument" should "return a dataset-typed LakeFSFileDocument for the dataset scheme" in { val datasetUri = new URI(s"dataset:///repo/$versionHash/file.txt") val doc = DocumentFactory.openReadonlyDocument(datasetUri) - doc shouldBe a[DatasetFileDocument] + doc shouldBe a[LakeFSFileDocument] + doc.asInstanceOf[LakeFSFileDocument].resourceType shouldBe ResourceType.Datasets doc.getURI shouldBe datasetUri } - it should "return a ModelFileDocument for the model scheme and parse its URI components" in { + it should "return a model-typed LakeFSFileDocument for the model scheme and parse its URI components" in { val modelUri = new URI(s"model:///model-1/$versionHash/weights/model.pt") val doc = DocumentFactory.openReadonlyDocument(modelUri) - doc shouldBe a[ModelFileDocument] + doc shouldBe a[LakeFSFileDocument] + doc.asInstanceOf[LakeFSFileDocument].resourceType shouldBe ResourceType.Models doc.getURI shouldBe modelUri val resource = doc.asInstanceOf[OnVersionedFileResource] @@ -176,17 +177,19 @@ class DocumentFactorySpec extends AnyFlatSpec with Matchers with BeforeAndAfterA // openDocument / createDocument / documentExists -- unsupported schemes // --------------------------------------------------------------------------- - "openDocument" should "return a DatasetFileDocument and no schema for the dataset scheme" in { + "openDocument" should "return a dataset-typed LakeFSFileDocument and no schema for the dataset scheme" in { val datasetUri = new URI(s"dataset:///repo/$versionHash/file.txt") val (doc, schemaOpt) = DocumentFactory.openDocument(datasetUri) - doc shouldBe a[DatasetFileDocument] + doc shouldBe a[LakeFSFileDocument] + doc.asInstanceOf[LakeFSFileDocument].resourceType shouldBe ResourceType.Datasets schemaOpt shouldBe None } - it should "return a ModelFileDocument and no schema for the model scheme" in { + it should "return a model-typed LakeFSFileDocument and no schema for the model scheme" in { val modelUri = new URI(s"model:///model-1/$versionHash/weights/model.pt") val (doc, schemaOpt) = DocumentFactory.openDocument(modelUri) - doc shouldBe a[ModelFileDocument] + doc shouldBe a[LakeFSFileDocument] + doc.asInstanceOf[LakeFSFileDocument].resourceType shouldBe ResourceType.Models schemaOpt shouldBe None } diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/model/DatasetFileDocumentSpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/model/LakeFSFileDocumentSpec.scala similarity index 77% rename from common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/model/DatasetFileDocumentSpec.scala rename to common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/model/LakeFSFileDocumentSpec.scala index 5a313c2763..56d5cd36ce 100644 --- a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/model/DatasetFileDocumentSpec.scala +++ b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/model/LakeFSFileDocumentSpec.scala @@ -20,6 +20,7 @@ package org.apache.texera.amber.core.storage.model import org.apache.texera.common.config.EnvironmentalVariable +import org.apache.texera.amber.core.storage.ResourceType import org.scalatest.flatspec.AnyFlatSpec import org.scalatest.matchers.should.Matchers @@ -27,15 +28,18 @@ import java.net.{URI, URLEncoder} import java.nio.charset.StandardCharsets import java.nio.file.Paths -class DatasetFileDocumentSpec extends AnyFlatSpec with Matchers { +class LakeFSFileDocumentSpec extends AnyFlatSpec with Matchers { // Realistic 40-char git commit hash, mirroring the URIs produced by FileResolver // (format: dataset:///{repositoryName}/{versionHash}/{fileRelativePath}). private val versionHash = "97fd4c2a755b69b7c66d322eab40b7e5c2ad5d10" - "DatasetFileDocument" should "parse a valid 3-segment dataset URI into its components" in { + // URI parsing is shared by every resource type; exercise it through the dataset type. + private def datasetDoc(uri: URI) = new LakeFSFileDocument(uri, ResourceType.Datasets) + + "LakeFSFileDocument" should "parse a valid 3-segment dataset URI into its components" in { val uri = new URI(s"dataset:///test_dataset/$versionHash/1.txt") - val doc = new DatasetFileDocument(uri) + val doc = datasetDoc(uri) doc.getRepositoryName() shouldBe "test_dataset" doc.getVersionHash() shouldBe versionHash @@ -44,7 +48,7 @@ class DatasetFileDocumentSpec extends AnyFlatSpec with Matchers { it should "join multi-segment relative paths correctly" in { val uri = new URI(s"dataset:///my_repo/$versionHash/some/nested/dir/data.csv") - val doc = new DatasetFileDocument(uri) + val doc = datasetDoc(uri) doc.getRepositoryName() shouldBe "my_repo" doc.getVersionHash() shouldBe versionHash @@ -55,7 +59,7 @@ class DatasetFileDocumentSpec extends AnyFlatSpec with Matchers { // FileResolver URL-encodes segments and then builds the URI with the multi-arg // constructor, so uri.getPath still contains URLEncoder-encoded segments. val uri = new URI("dataset", "", "/repo/hash%20with%2Bspecials/file.txt", null) - val doc = new DatasetFileDocument(uri) + val doc = datasetDoc(uri) doc.getVersionHash() shouldBe "hash with+specials" doc.getFileRelativePath() shouldBe "file.txt" @@ -63,7 +67,7 @@ class DatasetFileDocumentSpec extends AnyFlatSpec with Matchers { it should "URL-decode each relative path segment" in { val uri = new URI("dataset", "", "/repo/hash/dir+one/file%23two%20a.csv", null) - val doc = new DatasetFileDocument(uri) + val doc = datasetDoc(uri) doc.getRepositoryName() shouldBe "repo" doc.getVersionHash() shouldBe "hash" @@ -72,7 +76,7 @@ class DatasetFileDocumentSpec extends AnyFlatSpec with Matchers { it should "return the parsed components and the original URI through its getters" in { val uri = new URI("dataset", "", s"/repo/$versionHash/a%20b/c.csv", null) - val doc = new DatasetFileDocument(uri) + val doc = datasetDoc(uri) doc.getRepositoryName() shouldBe "repo" doc.getVersionHash() shouldBe versionHash @@ -89,7 +93,7 @@ class DatasetFileDocumentSpec extends AnyFlatSpec with Matchers { // URI constructor is required here: a single-arg URI already percent-decodes // getPath, so "%20" in a raw URI string would reach parseUri as a space. val uri = new URI("dataset", "", "/repo%20name/hash%20value/file.txt", null) - val doc = new DatasetFileDocument(uri) + val doc = datasetDoc(uri) doc.getRepositoryName() shouldBe "repo%20name" // Same encoded token in the version-hash position IS decoded (asymmetry pin). @@ -102,7 +106,7 @@ class DatasetFileDocumentSpec extends AnyFlatSpec with Matchers { val encodedPath = rawSegments.map(URLEncoder.encode(_, StandardCharsets.UTF_8)).mkString("/") val uri = new URI("dataset", "", s"/repo/$versionHash/$encodedPath", null) - val doc = new DatasetFileDocument(uri) + val doc = datasetDoc(uri) doc.getFileRelativePath() shouldBe Paths.get(rawSegments.head, rawSegments.tail: _*).toString } @@ -110,7 +114,7 @@ class DatasetFileDocumentSpec extends AnyFlatSpec with Matchers { it should "collapse redundant and trailing slashes in the URI path" in { // Paths.get collapses duplicate separators and ignores a trailing slash, // so this still yields exactly the three segments [repo, hash, file.txt]. - val doc = new DatasetFileDocument(new URI("dataset:///repo//hash///file.txt/")) + val doc = datasetDoc(new URI("dataset:///repo//hash///file.txt/")) doc.getRepositoryName() shouldBe "repo" doc.getVersionHash() shouldBe "hash" @@ -120,10 +124,10 @@ class DatasetFileDocumentSpec extends AnyFlatSpec with Matchers { it should "preserve dot segments in the relative path without normalization (current behavior)" in { // "." and ".." segments are kept verbatim (current behavior): the relative // path is passed downstream un-normalized, with no sanitization applied. - val parentDoc = new DatasetFileDocument(new URI("dataset:///repo/hash/../x.csv")) + val parentDoc = datasetDoc(new URI("dataset:///repo/hash/../x.csv")) parentDoc.getFileRelativePath() shouldBe Paths.get("..", "x.csv").toString - val dotDoc = new DatasetFileDocument(new URI("dataset:///repo/hash/./sub/../x.csv")) + val dotDoc = datasetDoc(new URI("dataset:///repo/hash/./sub/../x.csv")) dotDoc.getFileRelativePath() shouldBe Paths.get(".", "sub", "..", "x.csv").toString } @@ -136,7 +140,7 @@ class DatasetFileDocumentSpec extends AnyFlatSpec with Matchers { ) invalidUris.foreach { uri => val thrown = intercept[IllegalArgumentException] { - new DatasetFileDocument(uri) + datasetDoc(uri) } thrown.getMessage shouldBe "URI format is incorrect" } @@ -147,8 +151,8 @@ class DatasetFileDocumentSpec extends AnyFlatSpec with Matchers { // asInputStream needs to fetch a file; assert its fallback behavior without // requiring a live FileService or LakeFS. The check is guarded so it holds // regardless of whether the env override is present. - "DatasetFileDocument companion" should - "expose the default presigned-URL endpoint when the env override is absent" in { + "the companion" should + "resolve the dataset presigned-URL endpoint, defaulting when the env override is absent" in { val expected = sys.env .getOrElse( @@ -156,12 +160,26 @@ class DatasetFileDocumentSpec extends AnyFlatSpec with Matchers { "http://localhost:9092/api/dataset/presign-download"; ) .trim - DatasetFileDocument.fileServiceGetPresignURLEndpoint shouldBe expected + LakeFSFileDocument.presignEndpointOf(ResourceType.Datasets) shouldBe expected + } + + // Each resource type resolves its own endpoint: a dataset grant must not authorize a model + // file, so the two presign endpoints stay distinct. + it should "resolve the model presigned-URL endpoint, defaulting when the env override is absent" in { + val expected = + sys.env + .getOrElse( + EnvironmentalVariable.ENV_FILE_SERVICE_GET_MODEL_PRESIGNED_URL_ENDPOINT, + "http://localhost:9092/api/model/presign-download"; + ) + .trim + LakeFSFileDocument.presignEndpointOf(ResourceType.Models) shouldBe expected + LakeFSFileDocument.presignEndpointOf(ResourceType.Models) should not be + LakeFSFileDocument.presignEndpointOf(ResourceType.Datasets) } - // The user JWT token is shared by every LakeFS-backed document, so it now lives on - // the LakeFSFileDocument base object rather than on DatasetFileDocument. - "LakeFSFileDocument companion" should "expose a trimmed user JWT token defaulting to empty" in { + // The user JWT token is shared by every LakeFS-backed document, whatever its resource type. + it should "expose a trimmed user JWT token defaulting to empty" in { val expected = sys.env.getOrElse(EnvironmentalVariable.ENV_USER_JWT_TOKEN, "").trim LakeFSFileDocument.userJwtToken shouldBe expected diff --git a/file-service/src/main/scala/org/apache/texera/service/resource/DatasetAccessResource.scala b/file-service/src/main/scala/org/apache/texera/service/resource/DatasetAccessResource.scala index e03529fd7c..ad26338968 100644 --- a/file-service/src/main/scala/org/apache/texera/service/resource/DatasetAccessResource.scala +++ b/file-service/src/main/scala/org/apache/texera/service/resource/DatasetAccessResource.scala @@ -26,18 +26,11 @@ import jakarta.ws.rs._ import org.apache.texera.auth.SessionUser import org.apache.texera.dao.SqlServer import org.apache.texera.dao.SqlServer.withTransaction -import org.apache.texera.dao.jooq.generated.Tables.USER import org.apache.texera.dao.jooq.generated.enums.PrivilegeEnum -import org.apache.texera.dao.jooq.generated.tables.DatasetUserAccess.DATASET_USER_ACCESS -import org.apache.texera.dao.jooq.generated.tables.daos.{DatasetDao, DatasetUserAccessDao, UserDao} -import org.apache.texera.dao.jooq.generated.tables.pojos.{DatasetUserAccess, User} -import org.apache.texera.service.resource.DatasetAccessResource.{ - AccessEntry, - context, - getOwner, - userHasWriteAccess -} -import org.jooq.{DSLContext, EnumType} +import org.apache.texera.dao.jooq.generated.tables.pojos.User +import org.apache.texera.service.resource.DatasetAccessResource.context +import org.apache.texera.service.resource.ResourceTables.{Dataset => DATASET_RESOURCE} +import org.jooq.DSLContext object DatasetAccessResource { private def context: DSLContext = @@ -45,61 +38,29 @@ object DatasetAccessResource { .getInstance() .createDSLContext() - def isDatasetPublic(ctx: DSLContext, did: Integer): Boolean = { - val datasetDao = new DatasetDao(ctx.configuration()) - Option(datasetDao.fetchOneByDid(did)) - .flatMap(dataset => Option(dataset.getIsPublic)) - .contains(true) - } + type AccessEntry = ResourceAccess.AccessEntry + val AccessEntry: ResourceAccess.AccessEntry.type = ResourceAccess.AccessEntry - def userHasReadAccess(ctx: DSLContext, did: Integer, uid: Integer): Boolean = { - isDatasetPublic(ctx, did) || - userHasWriteAccess(ctx, did, uid) || - getDatasetUserAccessPrivilege(ctx, did, uid) == PrivilegeEnum.READ - } + def isDatasetPublic(ctx: DSLContext, did: Integer): Boolean = + ResourceAccess.isPublic(ctx, DATASET_RESOURCE, did) - def userOwnDataset(ctx: DSLContext, did: Integer, uid: Integer): Boolean = { - val datasetDao = new DatasetDao(ctx.configuration()) + def userHasReadAccess(ctx: DSLContext, did: Integer, uid: Integer): Boolean = + ResourceAccess.userHasReadAccess(ctx, DATASET_RESOURCE, did, uid) - Option(datasetDao.fetchOneByDid(did)) - .exists(_.getOwnerUid == uid) - } + def userOwnDataset(ctx: DSLContext, did: Integer, uid: Integer): Boolean = + ResourceAccess.userOwns(ctx, DATASET_RESOURCE, did, uid) - def userHasWriteAccess(ctx: DSLContext, did: Integer, uid: Integer): Boolean = { - userOwnDataset(ctx, did, uid) || - getDatasetUserAccessPrivilege(ctx, did, uid) == PrivilegeEnum.WRITE - } + def userHasWriteAccess(ctx: DSLContext, did: Integer, uid: Integer): Boolean = + ResourceAccess.userHasWriteAccess(ctx, DATASET_RESOURCE, did, uid) def getDatasetUserAccessPrivilege( ctx: DSLContext, did: Integer, uid: Integer - ): PrivilegeEnum = { - Option( - ctx - .select(DATASET_USER_ACCESS.PRIVILEGE) - .from(DATASET_USER_ACCESS) - .where( - DATASET_USER_ACCESS.DID - .eq(did) - .and(DATASET_USER_ACCESS.UID.eq(uid)) - ) - .fetchOneInto(classOf[PrivilegeEnum]) - ).getOrElse(PrivilegeEnum.NONE) - } - - def getOwner(ctx: DSLContext, did: Integer): User = { - val datasetDao = new DatasetDao(ctx.configuration()) - val userDao = new UserDao(ctx.configuration()) - - Option(datasetDao.fetchOneByDid(did)) - .flatMap(dataset => Option(dataset.getOwnerUid)) - .map(ownerUid => userDao.fetchOneByUid(ownerUid)) - .orNull - } - - case class AccessEntry(email: String, name: String, privilege: EnumType) {} + ): PrivilegeEnum = ResourceAccess.privilegeOf(ctx, DATASET_RESOURCE, did, uid) + def getOwner(ctx: DSLContext, did: Integer): User = + ResourceAccess.owner(ctx, DATASET_RESOURCE, did) } @Produces(Array(MediaType.APPLICATION_JSON)) @@ -115,16 +76,13 @@ class DatasetAccessResource { */ @GET @Path("/owner/{did}") - def getOwnerEmailOfDataset(@PathParam("did") did: Integer): String = { - var email = "" - withTransaction(context) { ctx => - val owner = getOwner(ctx, did) - if (owner != null) { - email = owner.getEmail - } - } - email - } + def getOwnerEmailOfDataset( + @PathParam("did") did: Integer, + @Auth user: SessionUser + ): String = + withTransaction(context)(ctx => + ResourceAccess.ownerEmail(ctx, DATASET_RESOURCE, did, user.getUid) + ) /** * Returns information about all current shared access of the given dataset @@ -135,27 +93,12 @@ class DatasetAccessResource { @GET @Path("/list/{did}") def getAccessList( - @PathParam("did") did: Integer - ): java.util.List[AccessEntry] = { - withTransaction(context) { ctx => - val datasetDao = new DatasetDao(ctx.configuration()) - ctx - .select( - USER.EMAIL, - USER.NAME, - DATASET_USER_ACCESS.PRIVILEGE - ) - .from(DATASET_USER_ACCESS) - .join(USER) - .on(USER.UID.eq(DATASET_USER_ACCESS.UID)) - .where( - DATASET_USER_ACCESS.DID - .eq(did) - .and(DATASET_USER_ACCESS.UID.notEqual(datasetDao.fetchOneByDid(did).getOwnerUid)) - ) - .fetchInto(classOf[AccessEntry]) - } - } + @PathParam("did") did: Integer, + @Auth user: SessionUser + ): java.util.List[DatasetAccessResource.AccessEntry] = + withTransaction(context)(ctx => + ResourceAccess.accessList(ctx, DATASET_RESOURCE, did, user.getUid) + ) /** * This method shares a dataset to a user with a specific access type @@ -172,27 +115,10 @@ class DatasetAccessResource { @PathParam("email") email: String, @PathParam("privilege") privilege: String, @Auth user: SessionUser - ): Response = { + ): Response = withTransaction(context) { ctx => - if (!userHasWriteAccess(ctx, did, user.getUid)) { - throw new ForbiddenException(s"You do not have permission to modify dataset $did") - } - val datasetUserAccessDao = new DatasetUserAccessDao(ctx.configuration()) - val userDao = new UserDao(ctx.configuration()) - val targetUser = userDao.fetchOneByEmail(email) - if (targetUser == null || targetUser.getIsPlaceholder) { - throw new BadRequestException(s"No registered user with email $email") - } - datasetUserAccessDao.merge( - new DatasetUserAccess( - did, - targetUser.getUid, - PrivilegeEnum.valueOf(privilege) - ) - ) - Response.ok().build() + ResourceAccess.grant(ctx, DATASET_RESOURCE, did, email, privilege, user.getUid) } - } /** * This method revoke the user's access of the given dataset @@ -207,24 +133,8 @@ class DatasetAccessResource { @PathParam("did") did: Integer, @PathParam("email") email: String, @Auth user: SessionUser - ): Response = { + ): Response = withTransaction(context) { ctx => - if (!userHasWriteAccess(ctx, did, user.getUid)) { - throw new ForbiddenException(s"You do not have permission to modify dataset $did") - } - - val userDao = new UserDao(ctx.configuration()) - - ctx - .delete(DATASET_USER_ACCESS) - .where( - DATASET_USER_ACCESS.UID - .eq(userDao.fetchOneByEmail(email).getUid) - .and(DATASET_USER_ACCESS.DID.eq(did)) - ) - .execute() - - Response.ok().build() + ResourceAccess.revoke(ctx, DATASET_RESOURCE, did, email, user.getUid) } - } } diff --git a/file-service/src/main/scala/org/apache/texera/service/resource/DatasetResource.scala b/file-service/src/main/scala/org/apache/texera/service/resource/DatasetResource.scala index 75391149c6..8bb1a70661 100644 --- a/file-service/src/main/scala/org/apache/texera/service/resource/DatasetResource.scala +++ b/file-service/src/main/scala/org/apache/texera/service/resource/DatasetResource.scala @@ -51,6 +51,7 @@ import org.apache.texera.dao.jooq.generated.tables.pojos.{ } import org.apache.texera.service.`type`.DatasetFileNode import org.apache.texera.service.resource.DatasetAccessResource._ +import org.apache.texera.service.resource.ResourceTables.{Dataset => DATASET_RESOURCE} import org.apache.texera.service.resource.DatasetResource.{context, _} import org.apache.texera.service.util.S3StorageClient import org.apache.texera.service.util.S3StorageClient.{ @@ -407,17 +408,7 @@ class DatasetResource extends LazyLogging { val isDatasetDownloadable = request.isDatasetDownloadable validateDatasetName(datasetName) - - // Check if a dataset with the same name already exists - val duplicateExists = ctx.fetchExists( - ctx - .selectFrom(DATASET) - .where(DATASET.OWNER_UID.eq(uid)) - .and(DATASET.NAME.eq(datasetName)) - ) - if (duplicateExists) { - throw new BadRequestException("Dataset with the same name already exists") - } + ResourceNaming.requireNameAvailable(ctx, DATASET_RESOURCE, uid, datasetName) // insert the dataset into the database val dataset = new Dataset() @@ -672,18 +663,13 @@ class DatasetResource extends LazyLogging { } validateDatasetName(modificator.name) - - // Check if the owner already has another dataset with the same name - val duplicateExists = ctx.fetchExists( - ctx - .selectFrom(DATASET) - .where(DATASET.OWNER_UID.eq(dataset.getOwnerUid)) - .and(DATASET.NAME.eq(modificator.name)) - .and(DATASET.DID.notEqual(dataset.getDid)) + ResourceNaming.requireNameAvailable( + ctx, + DATASET_RESOURCE, + dataset.getOwnerUid, + modificator.name, + excludingId = Some(dataset.getDid) ) - if (duplicateExists) { - throw new BadRequestException("Dataset with the same name already exists") - } dataset.setName(modificator.name) failOnDuplicateDatasetName { @@ -1305,49 +1291,24 @@ class DatasetResource extends LazyLogging { ): List[DashboardDataset] = { val uid = user.getUid withTransaction(context)(ctx => { - var accessibleDatasets: ListBuffer[DashboardDataset] = ListBuffer() - // first fetch all datasets user have explicit access to - accessibleDatasets = ListBuffer.from( - ctx - .select() - .from( - DATASET - .leftJoin(DATASET_USER_ACCESS) - .on(DATASET_USER_ACCESS.DID.eq(DATASET.DID)) - .leftJoin(USER) - .on(USER.UID.eq(DATASET.OWNER_UID)) - ) - .where(DATASET_USER_ACCESS.UID.eq(uid)) - .fetch() - .map(record => { - val dataset = record.into(DATASET).into(classOf[Dataset]) - val datasetAccess = record.into(DATASET_USER_ACCESS).into(classOf[DatasetUserAccess]) - val ownerEmail = record.into(USER).getEmail + ResourceAccess.listVisible( + ctx, + DATASET_RESOURCE, + uid, + classOf[Dataset], + (dataset: Dataset) => dataset.getDid + )( + fromGrant = (dataset, ownerEmail, privilege, isOwner) => + Some( DashboardDataset( - isOwner = dataset.getOwnerUid == uid, + isOwner = isOwner, dataset = dataset, - accessPrivilege = datasetAccess.getPrivilege, + accessPrivilege = privilege, ownerEmail = ownerEmail, size = 0 ) - }) - .asScala - ) - - // then we fetch the public datasets and merge it as a part of the result if not exist - val publicDatasets = ctx - .select() - .from( - DATASET - .leftJoin(USER) - .on(USER.UID.eq(DATASET.OWNER_UID)) - ) - .where(DATASET.IS_PUBLIC.eq(true)) - .fetch() - .asScala - .flatMap { record => - val dataset = record.into(DATASET).into(classOf[Dataset]) - val ownerEmail = record.into(USER).getEmail + ), + fromPublic = (dataset, ownerEmail) => try { Some( DashboardDataset( @@ -1366,13 +1327,7 @@ class DatasetResource extends LazyLogging { ) None } - } - publicDatasets.foreach { publicDataset => - if (!accessibleDatasets.exists(_.dataset.getDid == publicDataset.dataset.getDid)) { - accessibleDatasets = accessibleDatasets :+ publicDataset - } - } - accessibleDatasets.toList + ) }) } @@ -1606,47 +1561,13 @@ class DatasetResource extends LazyLogging { .fetchInto(classOf[String]) } - private val DATASET_NAME_MAX_LENGTH = 128 - private val DATASET_NAME_PATTERN = "^[A-Za-z0-9_-]+$".r + /** @see [[ResourceNaming.validateName]] */ + private def validateDatasetName(name: String): Unit = + ResourceNaming.validateName(DATASET_RESOURCE.label, name) - /** - * Validates the dataset name. - * - * Rules: - * - Must be 1 to 128 characters long. - * - Only letters, numbers, underscores, and hyphens are allowed. - * - * @param name The dataset name to validate. - * @throws jakarta.ws.rs.BadRequestException if the name is invalid. - */ - private def validateDatasetName(name: String): Unit = { - if (name == null || !DATASET_NAME_PATTERN.matches(name)) { - throw new BadRequestException( - "Invalid dataset name: only letters, numbers, underscores, and hyphens are allowed." - ) - } - if (name.length > DATASET_NAME_MAX_LENGTH) { - throw new BadRequestException( - s"Invalid dataset name: name must be at most $DATASET_NAME_MAX_LENGTH characters long." - ) - } - } - - /** - * Runs a dataset write and translates a (owner_uid, name) unique-constraint - * violation into the same BadRequestException the pre-checks throw, so - * requests losing a concurrent race get a 400 instead of a 500. - */ - private[resource] def failOnDuplicateDatasetName[T](op: => T): T = { - try op - catch { - case e: DataAccessException => - if (e.sqlState() == "23505") { - throw new BadRequestException("Dataset with the same name already exists") - } - throw e - } - } + /** @see [[ResourceNaming.failOnDuplicateName]] */ + private[resource] def failOnDuplicateDatasetName[T](op: => T): T = + ResourceNaming.failOnDuplicateName(DATASET_RESOURCE.label)(op) private def fetchDatasetVersions(ctx: DSLContext, did: Integer): List[DatasetVersion] = { ctx diff --git a/file-service/src/main/scala/org/apache/texera/service/resource/ResourceAccess.scala b/file-service/src/main/scala/org/apache/texera/service/resource/ResourceAccess.scala new file mode 100644 index 0000000000..6fffd1d335 --- /dev/null +++ b/file-service/src/main/scala/org/apache/texera/service/resource/ResourceAccess.scala @@ -0,0 +1,321 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.texera.service.resource + +import jakarta.ws.rs.{BadRequestException, ForbiddenException} +import jakarta.ws.rs.core.Response +import org.apache.texera.dao.jooq.generated.Tables.USER +import org.apache.texera.dao.jooq.generated.enums.PrivilegeEnum +import org.apache.texera.dao.jooq.generated.tables.daos.UserDao +import org.apache.texera.dao.jooq.generated.tables.pojos.User +import org.jooq.{DSLContext, EnumType, Record} + +import scala.jdk.CollectionConverters._ + +/** + * Ownership and privilege rules shared by every access-controlled resource. + * + * A resource is readable when it is public, or the caller owns it, or the caller holds an + * explicit grant; it is writable when the caller owns it or holds a WRITE grant. + */ +object ResourceAccess { + + /** One shared grant, as returned by the access-list endpoints. */ + case class AccessEntry(email: String, name: String, privilege: EnumType) {} + + def isPublic[R <: Record, A <: Record]( + ctx: DSLContext, + resource: ResourceTables[R, A], + id: Integer + ): Boolean = + Option( + ctx + .select(resource.isPublicField) + .from(resource.table) + .where(resource.idField.eq(id)) + .fetchOne() + ).flatMap(record => Option(record.value1())) + .exists(_.booleanValue()) + + def userOwns[R <: Record, A <: Record]( + ctx: DSLContext, + resource: ResourceTables[R, A], + id: Integer, + uid: Integer + ): Boolean = + Option( + ctx + .select(resource.ownerUidField) + .from(resource.table) + .where(resource.idField.eq(id)) + .fetchOne() + ).flatMap(record => Option(record.value1())) + .contains(uid) + + def privilegeOf[R <: Record, A <: Record]( + ctx: DSLContext, + resource: ResourceTables[R, A], + id: Integer, + uid: Integer + ): PrivilegeEnum = + Option( + ctx + .select(resource.privilegeField) + .from(resource.accessTable) + .where( + resource.accessIdField + .eq(id) + .and(resource.accessUidField.eq(uid)) + ) + .fetchOneInto(classOf[PrivilegeEnum]) + ).getOrElse(PrivilegeEnum.NONE) + + def userHasWriteAccess[R <: Record, A <: Record]( + ctx: DSLContext, + resource: ResourceTables[R, A], + id: Integer, + uid: Integer + ): Boolean = + userOwns(ctx, resource, id, uid) || + privilegeOf(ctx, resource, id, uid) == PrivilegeEnum.WRITE + + def userHasReadAccess[R <: Record, A <: Record]( + ctx: DSLContext, + resource: ResourceTables[R, A], + id: Integer, + uid: Integer + ): Boolean = + isPublic(ctx, resource, id) || + userHasWriteAccess(ctx, resource, id, uid) || + privilegeOf(ctx, resource, id, uid) == PrivilegeEnum.READ + + /** The owning user, or null when the resource does not exist. */ + def owner[R <: Record, A <: Record]( + ctx: DSLContext, + resource: ResourceTables[R, A], + id: Integer + ): User = { + val userDao = new UserDao(ctx.configuration()) + Option( + ctx + .select(resource.ownerUidField) + .from(resource.table) + .where(resource.idField.eq(id)) + .fetchOne() + ).flatMap(record => Option(record.value1())) + .map(ownerUid => userDao.fetchOneByUid(ownerUid)) + .orNull + } + + /** + * The owner's email. + * + * @throws jakarta.ws.rs.ForbiddenException if the caller cannot read the resource. + */ + def ownerEmail[R <: Record, A <: Record]( + ctx: DSLContext, + resource: ResourceTables[R, A], + id: Integer, + requesterUid: Integer + ): String = { + requireReadAccess(ctx, resource, id, requesterUid) + Option(owner(ctx, resource, id)).map(_.getEmail).getOrElse("") + } + + /** + * Everyone the resource is shared with, excluding the owner's own row. + * + * Read access is required rather than write, + * + * @throws jakarta.ws.rs.ForbiddenException if the caller cannot read the resource. + */ + def accessList[R <: Record, A <: Record]( + ctx: DSLContext, + resource: ResourceTables[R, A], + id: Integer, + requesterUid: Integer + ): java.util.List[AccessEntry] = { + requireReadAccess(ctx, resource, id, requesterUid) + val ownerUid = ctx + .select(resource.ownerUidField) + .from(resource.table) + .where(resource.idField.eq(id)) + .fetchOne() + .value1() + + ctx + .select(USER.EMAIL, USER.NAME, resource.privilegeField) + .from(resource.accessTable) + .join(USER) + .on(USER.UID.eq(resource.accessUidField)) + .where( + resource.accessIdField + .eq(id) + .and(resource.accessUidField.notEqual(ownerUid)) + ) + .fetchInto(classOf[AccessEntry]) + } + + /** + * Every resource of this type the user may see: the ones they hold an explicit grant on, plus + * every public one, with public entries dropped when they duplicate a granted entry. + * + * @param pojoClass the generated POJO the resource table maps into + * @param idOf reads the resource's id, used to de-duplicate the two passes + * @param fromGrant builds an entry the user has an explicit grant on + * @param fromPublic builds an entry visible only because the resource is public + */ + def listVisible[R <: Record, A <: Record, P, D]( + ctx: DSLContext, + resource: ResourceTables[R, A], + uid: Integer, + pojoClass: Class[P], + idOf: P => Integer + )( + fromGrant: (P, String, PrivilegeEnum, Boolean) => Option[D], + fromPublic: (P, String) => Option[D] + ): List[D] = { + // (id, entry) pairs so the public pass can skip ids already granted, without re-querying + val granted: List[(Integer, D)] = ctx + .select() + .from( + resource.table + .leftJoin(resource.accessTable) + .on(resource.accessIdField.eq(resource.idField)) + .leftJoin(USER) + .on(USER.UID.eq(resource.ownerUidField)) + ) + .where(resource.accessUidField.eq(uid)) + .fetch() + .asScala + .toList + .flatMap { record => + val entity = record.into(resource.table).into(pojoClass) + val privilege = record.into(resource.accessTable).get(resource.privilegeField) + val isOwner = record.into(resource.table).get(resource.ownerUidField) == uid + fromGrant(entity, record.into(USER).getEmail, privilege, isOwner) + .map(entry => (idOf(entity), entry)) + } + + val grantedIds = granted.map(_._1).toSet + + val public = ctx + .select() + .from( + resource.table + .leftJoin(USER) + .on(USER.UID.eq(resource.ownerUidField)) + ) + .where(resource.isPublicField.eq(true)) + .fetch() + .asScala + .toList + .flatMap { record => + val entity = record.into(resource.table).into(pojoClass) + if (grantedIds.contains(idOf(entity))) None + else fromPublic(entity, record.into(USER).getEmail) + } + + granted.map(_._2) ++ public + } + + /** + * Grants `privilege` to the user with `email`, replacing any privilege they already hold. + */ + def grant[R <: Record, A <: Record]( + ctx: DSLContext, + resource: ResourceTables[R, A], + id: Integer, + email: String, + privilege: String, + requesterUid: Integer + ): Response = { + requireWriteAccess(ctx, resource, id, requesterUid) + val grantee = new UserDao(ctx.configuration()).fetchOneByEmail(email) + if (grantee == null || grantee.getIsPlaceholder) { + throw new BadRequestException(s"No registered user with email $email") + } + val granteeUid = grantee.getUid + val granted = PrivilegeEnum.valueOf(privilege) + + ctx + .insertInto(resource.accessTable) + .set(resource.accessIdField, id) + .set(resource.accessUidField, granteeUid) + .set(resource.privilegeField, granted) + .onConflict(resource.accessIdField, resource.accessUidField) + .doUpdate() + .set(resource.privilegeField, granted) + .execute() + + Response.ok().build() + } + + /** + * Removes the user's explicit grant; a no-op when they hold none. + * + * @throws jakarta.ws.rs.ForbiddenException if the caller cannot modify the resource. + */ + def revoke[R <: Record, A <: Record]( + ctx: DSLContext, + resource: ResourceTables[R, A], + id: Integer, + email: String, + requesterUid: Integer + ): Response = { + requireWriteAccess(ctx, resource, id, requesterUid) + val granteeUid = new UserDao(ctx.configuration()).fetchOneByEmail(email).getUid + + ctx + .delete(resource.accessTable) + .where( + resource.accessUidField + .eq(granteeUid) + .and(resource.accessIdField.eq(id)) + ) + .execute() + + Response.ok().build() + } + + private def requireWriteAccess[R <: Record, A <: Record]( + ctx: DSLContext, + resource: ResourceTables[R, A], + id: Integer, + uid: Integer + ): Unit = + if (!userHasWriteAccess(ctx, resource, id, uid)) { + throw new ForbiddenException( + s"You do not have permission to modify ${resource.label} $id" + ) + } + + private def requireReadAccess[R <: Record, A <: Record]( + ctx: DSLContext, + resource: ResourceTables[R, A], + id: Integer, + uid: Integer + ): Unit = + if (!userHasReadAccess(ctx, resource, id, uid)) { + throw new ForbiddenException( + s"You do not have access to ${resource.label} $id" + ) + } +} diff --git a/file-service/src/main/scala/org/apache/texera/service/resource/ResourceNaming.scala b/file-service/src/main/scala/org/apache/texera/service/resource/ResourceNaming.scala new file mode 100644 index 0000000000..269cbce917 --- /dev/null +++ b/file-service/src/main/scala/org/apache/texera/service/resource/ResourceNaming.scala @@ -0,0 +1,99 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.texera.service.resource + +import jakarta.ws.rs.BadRequestException +import org.apache.texera.dao.SqlStates +import org.jooq.{DSLContext, Record} +import org.jooq.exception.DataAccessException + +/** + * Naming rules shared by every user-owned resource: what a name may contain, and that a name + * is unique among one owner's resources of that type. + */ +object ResourceNaming { + + private val NAME_MAX_LENGTH = 128 + private val NAME_PATTERN = "^[A-Za-z0-9_-]+$".r + + /** + * Rules: + * - Must be 1 to 128 characters long. + * - Only letters, numbers, underscores, and hyphens are allowed. + * + * @throws jakarta.ws.rs.BadRequestException if the name is invalid. + */ + def validateName(label: String, name: String): Unit = { + if (name == null || !NAME_PATTERN.matches(name)) { + throw new BadRequestException( + s"Invalid $label name: only letters, numbers, underscores, and hyphens are allowed." + ) + } + if (name.length > NAME_MAX_LENGTH) { + throw new BadRequestException( + s"Invalid $label name: name must be at most $NAME_MAX_LENGTH characters long." + ) + } + } + + /** + * Rejects a name the owner already uses for another resource of the same type. + * + * @param excludingId the resource being renamed, so it does not conflict with itself + */ + def requireNameAvailable[R <: Record, A <: Record]( + ctx: DSLContext, + resource: ResourceTables[R, A], + ownerUid: Integer, + name: String, + excludingId: Option[Integer] = None + ): Unit = { + val taken = ctx.fetchExists( + excludingId.foldLeft( + ctx + .selectFrom(resource.table) + .where(resource.ownerUidField.eq(ownerUid)) + .and(resource.nameField.eq(name)) + )((query, id) => query.and(resource.idField.notEqual(id))) + ) + if (taken) { + throw duplicateName(resource.label) + } + } + + /** + * Runs a write and translates an (owner_uid, name) unique-constraint violation into the same + * BadRequestException the pre-check throws, so requests losing a concurrent race get a 400 + * instead of a 500. + */ + def failOnDuplicateName[T](label: String)(op: => T): T = { + try op + catch { + case e: DataAccessException => + if (e.sqlState() == SqlStates.UNIQUE_VIOLATION) { + throw duplicateName(label) + } + throw e + } + } + + private def duplicateName(label: String): BadRequestException = + new BadRequestException(s"${label.capitalize} with the same name already exists") +} diff --git a/file-service/src/main/scala/org/apache/texera/service/resource/ResourceTables.scala b/file-service/src/main/scala/org/apache/texera/service/resource/ResourceTables.scala new file mode 100644 index 0000000000..605ed934c9 --- /dev/null +++ b/file-service/src/main/scala/org/apache/texera/service/resource/ResourceTables.scala @@ -0,0 +1,69 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.texera.service.resource + +import org.apache.texera.dao.jooq.generated.enums.PrivilegeEnum +import org.apache.texera.dao.jooq.generated.tables.Dataset.DATASET +import org.apache.texera.dao.jooq.generated.tables.DatasetUserAccess.DATASET_USER_ACCESS +import org.apache.texera.dao.jooq.generated.tables.records.{DatasetRecord, DatasetUserAccessRecord} +import org.jooq.{Record, Table, TableField} + +/** + * The tables and columns that define one resource type, so the rules in [[ResourceAccess]] and + * [[ResourceNaming]] can be written once and told which columns to read. + * + * A resource fits only if its own table carries the owner, the name and the `is_public` flag, and + * it has a companion `*_user_access` table — today that is `dataset` and `model`. This is not a + * model of every shareable resource: `workflow` keeps ownership in `workflow_of_user`, and + * neither `project` nor `workflow_computing_unit` has `is_public`. + * + * @param label how the resource is named in user-facing messages ("dataset", "model") + * @tparam R record type of the resource table + * @tparam A record type of the companion user-access table + */ +case class ResourceTables[R <: Record, A <: Record]( + label: String, + idField: TableField[R, Integer], + ownerUidField: TableField[R, Integer], + nameField: TableField[R, String], + isPublicField: TableField[R, java.lang.Boolean], + accessIdField: TableField[A, Integer], + accessUidField: TableField[A, Integer], + privilegeField: TableField[A, PrivilegeEnum] +) { + def table: Table[R] = idField.getTable + def accessTable: Table[A] = accessIdField.getTable +} + +object ResourceTables { + + val Dataset: ResourceTables[DatasetRecord, DatasetUserAccessRecord] = + ResourceTables( + label = "dataset", + idField = DATASET.DID, + ownerUidField = DATASET.OWNER_UID, + nameField = DATASET.NAME, + isPublicField = DATASET.IS_PUBLIC, + accessIdField = DATASET_USER_ACCESS.DID, + accessUidField = DATASET_USER_ACCESS.UID, + privilegeField = DATASET_USER_ACCESS.PRIVILEGE + ) + +} diff --git a/file-service/src/test/scala/org/apache/texera/service/resource/DatasetAccessResourceSpec.scala b/file-service/src/test/scala/org/apache/texera/service/resource/DatasetAccessResourceSpec.scala index f73c7adf64..709ede7184 100644 --- a/file-service/src/test/scala/org/apache/texera/service/resource/DatasetAccessResourceSpec.scala +++ b/file-service/src/test/scala/org/apache/texera/service/resource/DatasetAccessResourceSpec.scala @@ -113,8 +113,11 @@ class DatasetAccessResourceSpec .insert(new DatasetUserAccess(did, uid, privilege)) } - private def accessList(did: Integer): List[DatasetAccessResource.AccessEntry] = - accessResource.getAccessList(did).asScala.toList + private def accessList( + did: Integer, + user: SessionUser = ownerSession + ): List[DatasetAccessResource.AccessEntry] = + accessResource.getAccessList(did, user).asScala.toList override protected def beforeAll(): Unit = { super.beforeAll() @@ -436,10 +439,57 @@ class DatasetAccessResourceSpec // =========================================================================== "getOwnerEmailOfDataset" should "return the owner's email" in { - accessResource.getOwnerEmailOfDataset(privateDataset.getDid) shouldEqual ownerUser.getEmail + accessResource.getOwnerEmailOfDataset( + privateDataset.getDid, + ownerSession + ) shouldEqual ownerUser.getEmail + } + + it should "be readable by a READ grantee" in { + grantDirectly(privateDataset.getDid, readGranteeUser.getUid, PrivilegeEnum.READ) + + accessResource.getOwnerEmailOfDataset( + privateDataset.getDid, + readGranteeSession + ) shouldEqual ownerUser.getEmail + } + + it should "be forbidden for a user with no access to a private dataset" in { + assertThrows[ForbiddenException] { + accessResource.getOwnerEmailOfDataset(privateDataset.getDid, strangerSession) + } + } + + it should "be readable by anyone for a public dataset" in { + accessResource.getOwnerEmailOfDataset( + publicDataset.getDid, + strangerSession + ) shouldEqual ownerUser.getEmail + } + + it should "be forbidden for a nonexistent dataset" in { + assertThrows[ForbiddenException] { + accessResource.getOwnerEmailOfDataset(nonExistentDid, ownerSession) + } } - it should "return an empty string for a nonexistent dataset" in { - accessResource.getOwnerEmailOfDataset(nonExistentDid) shouldEqual "" + // =========================================================================== + // getAccessList -- read guard + // =========================================================================== + + "getAccessList" should "be forbidden for a user with no access to a private dataset" in { + grantDirectly(privateDataset.getDid, readGranteeUser.getUid, PrivilegeEnum.READ) + + assertThrows[ForbiddenException] { + accessList(privateDataset.getDid, strangerSession) + } + } + + it should "be readable by a READ grantee" in { + grantDirectly(privateDataset.getDid, readGranteeUser.getUid, PrivilegeEnum.READ) + + accessList(privateDataset.getDid, readGranteeSession).map(_.email) should contain( + readGranteeUser.getEmail + ) } }
