This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/texera.git
The following commit(s) were added to refs/heads/main by this push:
new dcec7f3601 feat(LakeFS-mount): authorize and perform a repository
mount (#6896)
dcec7f3601 is described below
commit dcec7f3601e61436857a619365f7a2c5963c7589
Author: ali risheh <[email protected]>
AuthorDate: Tue Sep 15 02:48:54 2026 +0000
feat(LakeFS-mount): authorize and perform a repository mount (#6896)
### What changes were proposed in this PR?
Let a computing unit have a versioned LakeFS repository mounted into it,
on the infrastructure merged in #6866. This is the authorization step of
#6606: the endpoint that decides whether a mount may happen, and
performs it.
The per-node mounter is the one privileged component — root on every
node, listening on a hostPort — so it admits exactly one caller,
verified with `TokenReview` against a token bound to the
`texera-mounter` audience. Only access-control-service holds one, which
is also why it is the right place: it is already where the deployment
decides whether a user may act on a computing unit.
**The mounter authorizes nothing.** It performs what it is told, so
every decision has to be made before it is asked. The endpoint makes
four:
1. **The request has a shape a mount path can be built from** — each
component a single safe segment, so a separator, a `..` or a leading `-`
is refused by name rather than by whatever it fails next. The mounter
checks this too, being privileged; here it is checked first, before any
database work.
2. **The caller holds write access to the computing unit.** Mounting
puts data into someone's unit, so it takes the same privilege as any
other change to one; a read-only sharee may use the unit, not alter what
it can see.
3. **The caller may read the repository.** It is matched by name across
datasets and models — by name rather than by parsing an id out of it,
because an earlier migration backfilled that column from the dataset's
plain name — and anything other than exactly one match is refused rather
than resolved arbitrarily.
4. **The commit belongs to that repository**, so a well-formed but
unrelated digest cannot be mounted.
Only then does it resolve which node the unit's pod is on — itself,
rather than accepting one from the caller, or anything that can reach it
could aim requests at any node's privileged mounter — and forward.
file-service still re-checks read access on every byte it serves, but as
the last line rather than the only one: without the check here a caller
could have a mount created for a repository they cannot read, learning
that it exists and spending a node's resources on it. The read and
ownership rules move to a module both services share, so they decide
from one definition instead of two that can drift.
Mounts are released when the computing unit's pod is deleted, by the
mounter's pod watcher, so there is no unmount path. No new
configuration: the mounter's port and file-service's root are read from
the environment the chart already sets, so each value is written once,
in the chart. The whole feature stays behind `mounter.enabled`; with it
off the chart renders no new object and no new environment variable.
Nothing here is used only by the follow-up. Resolving a logical dataset
or model path to a repository and commit, and the engine client that
calls this endpoint, land in #6895 where they have callers.
### Any related issues, documentation, discussions?
Closes #6863 · part of #6606 · builds on #6866.
### How was this PR tested?
- New unit tests cover each refusal — no access to the unit, read-only
access to it, a repository the caller cannot read, one that does not
exist, a commit belonging to another repository, a malformed component,
a unit not yet scheduled, and mounting disabled or misconfigured — each
asserting the mounter was never asked. Plus the node lookup and the
mounter client's own request shape, service-account token and error
handling. Compile and tests green across the touched modules.
- Validated end to end on a single-node minikube with
`mounter.enabled=true`, deployed from this chart rather than by hand so
the wiring itself is what ran. Every call was made from inside an
unprivileged computing-unit pod: authorized mount `200`, the same mount
again `200`, unauthenticated `401`, a computing unit the caller may not
use `403`, a repository that does not exist `403`, a commit from another
repository `403`, a malformed name `400`, and calling the node mounter
directly `401` while its health probe stays open at `200`.
- The mount itself: `fuse.geesefs` at the expected path inside the pod,
file byte-exact, read-only, in a pod whose effective capabilities are
empty. The mounter's log confirms GeeseFS was pointed at file-service's
root.
- Terminating the computing unit took GeeseFS mounts from one to zero
and emptied the mount root, which is why there is no unmount endpoint.
- With the feature off, the service account, the pod-reader role, the
mounter objects, the projected token and every new environment variable
render zero times.
- The cluster run caught a real bug: file-service was being addressed
through an endpoint whose path had to be stripped back off, when what
GeeseFS needs is the root the S3 proxy is served at. Fixed.
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5
---------
Co-authored-by: Claude Opus 5 (1M context) <[email protected]>
---
.../texera/service/AccessControlService.scala | 2 +
.../resource/ComputingUnitMountResource.scala | 216 ++++++++++++++
.../service/util/ComputingUnitNodeLocator.scala | 114 ++++++++
.../apache/texera/service/util/MounterClient.scala | 157 ++++++++++
.../resource/ComputingUnitMountResourceSpec.scala | 322 +++++++++++++++++++++
.../util/ComputingUnitNodeLocatorSpec.scala | 66 +++++
.../texera/service/util/MounterClientSpec.scala | 153 ++++++++++
.../access-control-service-deployment.yaml | 12 +
.../access-control-service-service-account.yaml | 50 ++--
build.sbt | 2 +-
.../texera/service/resource/ResourceAccess.scala | 0
.../texera/service/resource/ResourceTables.scala | 0
12 files changed, 1076 insertions(+), 18 deletions(-)
diff --git
a/access-control-service/src/main/scala/org/apache/texera/service/AccessControlService.scala
b/access-control-service/src/main/scala/org/apache/texera/service/AccessControlService.scala
index 1f50c86c9f..ca2c797ab9 100644
---
a/access-control-service/src/main/scala/org/apache/texera/service/AccessControlService.scala
+++
b/access-control-service/src/main/scala/org/apache/texera/service/AccessControlService.scala
@@ -28,6 +28,7 @@ import org.apache.texera.dao.SqlServer
import org.apache.texera.service.activity.UserActivityEventListener
import org.apache.texera.service.resource.{
AccessControlResource,
+ ComputingUnitMountResource,
HealthCheckResource,
LiteLLMModelsResource,
LiteLLMProxyResource
@@ -68,6 +69,7 @@ class AccessControlService extends
Application[AccessControlServiceConfiguration
environment.jersey.register(classOf[AccessControlResource])
environment.jersey.register(classOf[LiteLLMProxyResource])
environment.jersey.register(classOf[LiteLLMModelsResource])
+ environment.jersey.register(classOf[ComputingUnitMountResource])
AuthFeatures.register(environment)
diff --git
a/access-control-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitMountResource.scala
b/access-control-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitMountResource.scala
new file mode 100644
index 0000000000..b521bdf821
--- /dev/null
+++
b/access-control-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitMountResource.scala
@@ -0,0 +1,216 @@
+/*
+ * 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 com.typesafe.scalalogging.LazyLogging
+import io.dropwizard.auth.Auth
+import jakarta.annotation.security.RolesAllowed
+import jakarta.ws.rs._
+import jakarta.ws.rs.core.MediaType
+import org.apache.texera.auth.JwtAuth.{jwtClaims, jwtToken}
+import org.apache.texera.auth.SessionUser
+import org.apache.texera.auth.util.ComputingUnitAccess
+import org.apache.texera.common.config.{EnvironmentalVariable,
KubernetesConfig}
+import org.apache.texera.dao.SqlServer
+import org.apache.texera.dao.SqlServer.withTransaction
+import org.apache.texera.dao.jooq.generated.Tables.{DATASET_VERSION,
MODEL_VERSION}
+import org.apache.texera.dao.jooq.generated.enums.PrivilegeEnum
+import org.apache.texera.dao.jooq.generated.tables.daos.{DatasetDao, ModelDao}
+import org.apache.texera.service.resource.ComputingUnitMountResource._
+import org.apache.texera.service.util.{
+ ComputingUnitNodeLocator,
+ MountRequestValidation,
+ MounterClient
+}
+
+import scala.jdk.CollectionConverters._
+
+/**
+ * The mount authority: this service decides whether a user may act on a
computing unit, so
+ * it is where a mount request is authorized before being forwarded to that
unit's node.
+ */
+@Path("/mounts")
+@RolesAllowed(Array("REGULAR", "ADMIN"))
+@Produces(Array(MediaType.APPLICATION_JSON))
+class ComputingUnitMountResource(
+ mounterEnabled: Boolean,
+ mounterPort: Option[Int],
+ fileServiceBaseUrl: Option[String],
+ nodeLocator: ComputingUnitNodeLocator,
+ mounter: MounterClient
+) extends LazyLogging {
+
+ // No-arg constructor for Jersey reflection. Tests use the param-ful form.
+ def this() =
+ this(
+ KubernetesConfig.mounterEnabled,
+ EnvironmentalVariable.get(MounterPortVariable).map(_.trim.toInt),
+ EnvironmentalVariable.get(FileServiceUrlVariable),
+ ComputingUnitNodeLocator,
+ MounterClient
+ )
+
+ @POST
+ @Path("/{cuid}")
+ @Consumes(Array(MediaType.APPLICATION_JSON))
+ def mount(
+ @PathParam("cuid") cuid: Int,
+ request: MountRequest,
+ @Auth user: SessionUser
+ ): MountInfo = {
+ val (port, fileService) = requireMountConfiguration()
+ try MountRequestValidation.validate(cuid.toString, request.repositoryName,
request.commitHash)
+ catch { case e: IllegalArgumentException => throw new
BadRequestException(e.getMessage) }
+ requireComputingUnitAccess(cuid, user)
+ requireRepositoryReadAccess(request.repositoryName, request.commitHash,
user.getUid)
+ val nodeIp = requireNodeIp(cuid)
+
+ // A token minted here, after the access check: GeeseFS keeps presenting
it for the life
+ // of the mount, so it must be one this service vouched for.
+ val mountPath =
+ try {
+ mounter.mount(
+ nodeIp,
+ port,
+ cuid.toString,
+ request.repositoryName,
+ request.commitHash,
+ jwtToken(jwtClaims(user.getUser)),
+ fileService
+ )
+ } catch {
+ case e: IllegalArgumentException =>
+ throw new BadRequestException(e.getMessage)
+ case e: MounterClient.MounterRequestException =>
+ logger.warn(s"node mounter at $nodeIp refused a mount for computing
unit $cuid", e)
+ throw new BadRequestException(e.getMessage)
+ }
+
+ logger.info(
+ s"user ${user.getUid} mounted
${request.repositoryName}:${request.commitHash} " +
+ s"onto computing unit $cuid at $mountPath"
+ )
+ MountInfo(request.repositoryName, request.commitHash, mountPath)
+ }
+
+ /**
+ * What a mount request needs beyond the request itself, both passed by the
chart. Missing
+ * with mounting off is said plainly, because the alternative is a
connection timeout to a
+ * node port nothing is listening on; missing with it on is a
misconfiguration, and naming
+ * the variable beats letting a half-formed request reach the mounter.
+ */
+ private def requireMountConfiguration(): (Int, String) = {
+ if (!mounterEnabled) {
+ throw new ServiceUnavailableException(
+ "Repository mounting is not enabled on this deployment."
+ )
+ }
+ def required[T](value: Option[T], variable: String): T =
+ value.getOrElse(
+ throw new InternalServerErrorException(
+ s"Repository mounting is enabled but $variable is unset."
+ )
+ )
+ (
+ required(mounterPort, MounterPortVariable),
+ required(fileServiceBaseUrl.filter(_.nonEmpty), FileServiceUrlVariable)
+ )
+ }
+
+ /**
+ * Mounting puts data into someone's computing unit, so it takes the same
privilege as any
+ * other change to one: ownership, or an explicit WRITE grant. A read-only
sharee may use
+ * the unit, not alter what it can see.
+ */
+ private def requireComputingUnitAccess(cuid: Int, user: SessionUser): Unit =
+ if (ComputingUnitAccess.getComputingUnitAccess(cuid, user.getUid) !=
PrivilegeEnum.WRITE) {
+ logger.warn(s"user ${user.getUid} denied mount access to computing unit
$cuid")
+ throw new ForbiddenException("No write access to this computing unit.")
+ }
+
+ /**
+ * The repository must be one the user may read, at a commit that belongs
to it.
+ *
+ * The mounter performs what it is told and authorizes nothing, so this is
where a mount is
+ * refused. file-service re-checks read access on every byte served through
its proxy, but
+ * that is the last line rather than this one: without the check here a
caller could have a
+ * mount created for a repository they cannot read, learning it exists and
spending a node's
+ * resources on it.
+ *
+ * A repository is matched by name rather than parsed, because
`sql/updates/15.sql`
+ * backfilled the column from the dataset's plain name; both resource kinds
are searched,
+ * and anything other than exactly one match is refused rather than
resolved arbitrarily.
+ */
+ private def requireRepositoryReadAccess(
+ repositoryName: String,
+ commitHash: String,
+ uid: Integer
+ ): Unit =
+ withTransaction(SqlServer.getInstance().createDSLContext()) { ctx =>
+ val datasets = new
DatasetDao(ctx.configuration()).fetchByRepositoryName(repositoryName)
+ val models = new
ModelDao(ctx.configuration()).fetchByRepositoryName(repositoryName)
+
+ val readable = (datasets.asScala.toList, models.asScala.toList) match {
+ case (dataset :: Nil, Nil) =>
+ ResourceAccess.userHasReadAccess(ctx, ResourceTables.Dataset,
dataset.getDid, uid) &&
+ ctx.fetchExists(
+ DATASET_VERSION,
+ DATASET_VERSION.DID
+ .eq(dataset.getDid)
+ .and(DATASET_VERSION.VERSION_HASH.eq(commitHash))
+ )
+ case (Nil, model :: Nil) =>
+ ResourceAccess.userHasReadAccess(ctx, ResourceTables.Model,
model.getMid, uid) &&
+ ctx.fetchExists(
+ MODEL_VERSION,
+
MODEL_VERSION.MID.eq(model.getMid).and(MODEL_VERSION.VERSION_HASH.eq(commitHash))
+ )
+ case _ => false
+ }
+
+ if (!readable) {
+ logger.warn(s"user $uid denied a mount of '$repositoryName' at
'$commitHash'")
+ throw new ForbiddenException("No read access to the requested
repository version.")
+ }
+ }
+
+ private def requireNodeIp(cuid: Int): String =
+ nodeLocator
+ .nodeIpOf(cuid)
+ .getOrElse(
+ throw new BadRequestException(
+ s"Computing unit $cuid is not running on a node yet; cannot manage
its mounts."
+ )
+ )
+}
+
+object ComputingUnitMountResource {
+
+ /** Set by the chart from the same value it gives the mounter DaemonSet's
hostPort. */
+ private val MounterPortVariable = "KUBERNETES_MOUNTER_PORT"
+
+ // file-service's root, which is what GeeseFS is pointed at: the S3 proxy is
served at the
+ // servlet root, so this is scheme and authority with no path.
+ private val FileServiceUrlVariable = "FILE_SERVICE_URL"
+
+ case class MountRequest(repositoryName: String, commitHash: String)
+
+ case class MountInfo(repositoryName: String, commitHash: String, mountPath:
String)
+}
diff --git
a/access-control-service/src/main/scala/org/apache/texera/service/util/ComputingUnitNodeLocator.scala
b/access-control-service/src/main/scala/org/apache/texera/service/util/ComputingUnitNodeLocator.scala
new file mode 100644
index 0000000000..237d6cb2c0
--- /dev/null
+++
b/access-control-service/src/main/scala/org/apache/texera/service/util/ComputingUnitNodeLocator.scala
@@ -0,0 +1,114 @@
+/*
+ * 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.util
+
+import com.fasterxml.jackson.databind.{JsonNode, ObjectMapper}
+import com.typesafe.scalalogging.LazyLogging
+import org.apache.texera.common.config.KubernetesConfig
+
+import java.io.FileInputStream
+import java.net.URI
+import java.net.http.{HttpClient, HttpRequest, HttpResponse}
+import java.nio.file.{Files, Paths}
+import java.security.KeyStore
+import java.security.cert.CertificateFactory
+import java.time.Duration
+import javax.net.ssl.{SSLContext, TrustManagerFactory}
+import scala.jdk.CollectionConverters._
+
+/**
+ * Finds the node a computing unit's pod runs on, so a mount can be sent to
that node's
+ * mounter.
+ *
+ * Resolved here rather than supplied by the caller: letting a caller name
the node would
+ * hand anything that can reach this service the ability to aim requests at
any node's
+ * privileged mounter.
+ */
+class ComputingUnitNodeLocator(fetchPod: String => Option[JsonNode]) extends
LazyLogging {
+
+ def nodeIpOf(cuid: Int): Option[String] = {
+ val podName = s"${KubernetesConfig.computeUnitPodNamePrefix}-$cuid"
+ fetchPod(podName).map(_.at("/status/hostIP").asText("")).filter(_.nonEmpty)
+ }
+}
+
+object ComputingUnitNodeLocator extends
ComputingUnitNodeLocator(InClusterKubernetesApi.getPod)
+
+// One read of one field, so the API is called directly rather than through a
Kubernetes
+// client library and its transitive dependencies.
+private[util] object InClusterKubernetesApi extends LazyLogging {
+
+ private val serviceAccountDir =
"/var/run/secrets/kubernetes.io/serviceaccount"
+ private val mapper = new ObjectMapper()
+
+ private lazy val client: HttpClient =
+ HttpClient
+ .newBuilder()
+ .connectTimeout(Duration.ofSeconds(5))
+ .sslContext(clusterSslContext)
+ .build()
+
+ /** Trusts only the cluster CA, so this talks to the API server and nothing
else. */
+ private def clusterSslContext: SSLContext = {
+ val certificates = {
+ val stream = new FileInputStream(s"$serviceAccountDir/ca.crt")
+ try
CertificateFactory.getInstance("X.509").generateCertificates(stream).asScala.toList
+ finally stream.close()
+ }
+ val keyStore = KeyStore.getInstance(KeyStore.getDefaultType)
+ keyStore.load(null, null)
+ certificates.zipWithIndex.foreach {
+ case (certificate, index) =>
keyStore.setCertificateEntry(s"cluster-ca-$index", certificate)
+ }
+ val trustManagerFactory =
+ TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm)
+ trustManagerFactory.init(keyStore)
+ val context = SSLContext.getInstance("TLS")
+ context.init(null, trustManagerFactory.getTrustManagers, null)
+ context
+ }
+
+ def getPod(podName: String): Option[JsonNode] = {
+ val host = sys.env.getOrElse("KUBERNETES_SERVICE_HOST",
"kubernetes.default.svc")
+ val port = sys.env.getOrElse("KUBERNETES_SERVICE_PORT", "443")
+ val namespace = KubernetesConfig.computeUnitPoolNamespace
+ val token = Files.readString(Paths.get(s"$serviceAccountDir/token")).trim
+
+ val request = HttpRequest
+ .newBuilder()
+
.uri(URI.create(s"https://$host:$port/api/v1/namespaces/$namespace/pods/$podName";))
+ .header("Authorization", s"Bearer $token")
+ .timeout(Duration.ofSeconds(10))
+ .GET()
+ .build()
+
+ val response = client.send(request, HttpResponse.BodyHandlers.ofString())
+ response.statusCode() match {
+ case 200 => Some(mapper.readTree(response.body()))
+ case 404 => None
+ case other =>
+ // Distinguished from a missing pod: a missing RBAC rule or an
unreachable API
+ // server must not read as "the computing unit is not running".
+ throw new IllegalStateException(
+ s"cannot read pod $podName in namespace $namespace: HTTP $other
${response.body()}"
+ )
+ }
+ }
+}
diff --git
a/access-control-service/src/main/scala/org/apache/texera/service/util/MounterClient.scala
b/access-control-service/src/main/scala/org/apache/texera/service/util/MounterClient.scala
new file mode 100644
index 0000000000..1cb15f0213
--- /dev/null
+++
b/access-control-service/src/main/scala/org/apache/texera/service/util/MounterClient.scala
@@ -0,0 +1,157 @@
+/*
+ * 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.util
+
+import com.fasterxml.jackson.databind.{JsonNode, ObjectMapper}
+import com.fasterxml.jackson.module.scala.DefaultScalaModule
+
+import java.net.{HttpURLConnection, URI}
+import java.nio.charset.StandardCharsets
+import java.nio.file.{Files, Paths}
+import scala.util.Using
+
+/**
+ * HTTP client for the per-node `texera-mounter`.
+ *
+ * The mounter is privileged and listens on a hostPort, so it admits exactly
one caller: it
+ * requires a service-account token minted for its own audience and checks it
with the
+ * Kubernetes TokenReview API (see `authenticate_caller` in
`bin/mounter/mounter.py`). That
+ * caller is this service, which is why the client lives here.
+ */
+class MounterClient(tokenPath: String = MounterDefaults.ProjectedTokenPath) {
+
+ import MounterClient._
+
+ private val mapper: ObjectMapper = new
ObjectMapper().registerModule(DefaultScalaModule)
+
+ private val connectTimeoutMs = 10000
+ private val readTimeoutMs = 35000
+
+ private def baseUrl(nodeIp: String, port: Int): String =
s"http://$nodeIp:$port";
+
+ // Read per call, not cached: the kubelet rewrites the projected token in
place.
+ private def mounterToken(): String =
+ try Files.readString(Paths.get(tokenPath)).trim
+ catch {
+ case e: Exception =>
+ throw new IllegalStateException(
+ s"cannot read the mounter service-account token at $tokenPath;
without it this " +
+ s"service cannot authenticate to the node mounter: ${e.getMessage}"
+ )
+ }
+
+ def mount(
+ nodeIp: String,
+ port: Int,
+ cuid: String,
+ repositoryName: String,
+ commitHash: String,
+ jwt: String,
+ fileServiceBase: String
+ ): String = {
+ MountRequestValidation.validate(cuid, repositoryName, commitHash)
+
+ val body = mapper.createObjectNode()
+ body.put("cuid", cuid)
+ body.put("repositoryName", repositoryName)
+ body.put("commitHash", commitHash)
+ body.put("jwt", jwt)
+ body.put("fileServiceBase", fileServiceBase)
+
+ send("POST", s"${baseUrl(nodeIp, port)}/mount", Some(body.toString))
+ }
+
+ private def send(method: String, url: String, body: Option[String]): String
= {
+ val connection =
URI.create(url).toURL.openConnection().asInstanceOf[HttpURLConnection]
+ connection.setRequestMethod(method)
+ connection.setRequestProperty("Authorization", s"Bearer ${mounterToken()}")
+ connection.setConnectTimeout(connectTimeoutMs)
+ connection.setReadTimeout(readTimeoutMs)
+ body.foreach { _ =>
+ connection.setRequestProperty("Content-Type", "application/json")
+ connection.setDoOutput(true)
+ }
+ try {
+ body.foreach(payload =>
+
Using(connection.getOutputStream)(_.write(payload.getBytes(StandardCharsets.UTF_8)))
+ )
+ val code = connection.getResponseCode
+ val stream =
+ if (code >= 200 && code < 300) connection.getInputStream else
connection.getErrorStream
+ val responseBody = Option(stream)
+ .map(s => new String(s.readAllBytes(), StandardCharsets.UTF_8))
+ .getOrElse("")
+ if (code < 200 || code >= 300) {
+ throw new MounterRequestException(code, s"mounter $method failed: HTTP
$code $responseBody")
+ }
+ val response: JsonNode =
+ if (responseBody.isEmpty) mapper.createObjectNode() else
mapper.readTree(responseBody)
+ Option(response.get("mountPath"))
+ .filter(node => node.isTextual && node.asText().nonEmpty)
+ .map(_.asText())
+ .getOrElse(
+ throw new IllegalStateException(
+ s"mounter $method reported success without a mount path:
$responseBody"
+ )
+ )
+ } finally {
+ connection.disconnect()
+ }
+ }
+}
+
+object MounterClient extends MounterClient(MounterDefaults.ProjectedTokenPath)
{
+
+ class MounterRequestException(val status: Int, message: String) extends
RuntimeException(message)
+}
+
+private object MounterDefaults {
+
+ /** Where the chart projects the token. Not configurable: the same chart
fixes both ends. */
+ val ProjectedTokenPath = "/var/run/secrets/texera/mounter/token"
+}
+
+/**
+ * The shape a mount request has to have before anything acts on it.
+ *
+ * The mounter joins these into a path and creates the directory, so each has
to be a single
+ * safe segment: no separator, no "..", and a leading alphanumeric so a value
cannot be read
+ * as a geesefs flag. The mounter enforces this itself, being privileged;
this is the same
+ * rule applied earlier, so a malformed request is refused by name rather
than by whatever
+ * it fails next.
+ */
+private[service] object MountRequestValidation {
+
+ private val cuidPattern = "^[0-9]+$".r
+ private val segmentPattern = "^[A-Za-z0-9][A-Za-z0-9._-]*$".r
+
+ def validate(cuid: String, repositoryName: String, commitHash: String): Unit
= {
+ if (cuid == null || cuidPattern.findFirstIn(cuid).isEmpty) {
+ throw new IllegalArgumentException(s"cuid must be a non-negative
integer, got '$cuid'")
+ }
+ requireSegment(repositoryName, "repositoryName")
+ requireSegment(commitHash, "commitHash")
+ }
+
+ private def requireSegment(value: String, field: String): Unit =
+ if (value == null || segmentPattern.findFirstIn(value).isEmpty) {
+ throw new IllegalArgumentException(s"$field must be a single path
segment, got '$value'")
+ }
+}
diff --git
a/access-control-service/src/test/scala/org/apache/texera/service/resource/ComputingUnitMountResourceSpec.scala
b/access-control-service/src/test/scala/org/apache/texera/service/resource/ComputingUnitMountResourceSpec.scala
new file mode 100644
index 0000000000..926568fc83
--- /dev/null
+++
b/access-control-service/src/test/scala/org/apache/texera/service/resource/ComputingUnitMountResourceSpec.scala
@@ -0,0 +1,322 @@
+/*
+ * 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,
+ InternalServerErrorException,
+ ServiceUnavailableException
+}
+import org.apache.texera.auth.SessionUser
+import org.apache.texera.dao.MockTexeraDB
+import org.apache.texera.dao.jooq.generated.enums.{
+ PrivilegeEnum,
+ UserRoleEnum,
+ WorkflowComputingUnitTypeEnum
+}
+import org.apache.texera.dao.jooq.generated.tables.daos.{
+ ComputingUnitUserAccessDao,
+ DatasetDao,
+ DatasetVersionDao,
+ UserDao,
+ WorkflowComputingUnitDao
+}
+import org.apache.texera.dao.jooq.generated.tables.pojos.{
+ ComputingUnitUserAccess,
+ Dataset,
+ DatasetVersion,
+ User,
+ WorkflowComputingUnit
+}
+import
org.apache.texera.service.resource.ComputingUnitMountResource.MountRequest
+import org.apache.texera.service.util.{ComputingUnitNodeLocator, MounterClient}
+import org.scalatest.BeforeAndAfterAll
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+import scala.collection.mutable
+
+class ComputingUnitMountResourceSpec
+ extends AnyFlatSpec
+ with Matchers
+ with BeforeAndAfterAll
+ with MockTexeraDB {
+
+ private val owner: User = {
+ val user = new User
+ user.setUid(1)
+ user.setName("owner")
+ user.setEmail("[email protected]")
+ user.setRole(UserRoleEnum.REGULAR)
+ user
+ }
+
+ private val reader: User = {
+ val user = new User
+ user.setUid(2)
+ user.setName("reader")
+ user.setEmail("[email protected]")
+ user.setRole(UserRoleEnum.REGULAR)
+ user
+ }
+
+ private val stranger: User = {
+ val user = new User
+ user.setUid(3)
+ user.setName("stranger")
+ user.setEmail("[email protected]")
+ user.setRole(UserRoleEnum.REGULAR)
+ user
+ }
+
+ private val ownedDataset: Dataset = {
+ val dataset = new Dataset
+ dataset.setDid(1)
+ dataset.setName("owned")
+ dataset.setDescription("")
+ dataset.setRepositoryName("dataset-1")
+ dataset.setOwnerUid(owner.getUid)
+ dataset.setIsPublic(false)
+ dataset
+ }
+
+ private val ownedVersion: DatasetVersion = {
+ val version = new DatasetVersion
+ version.setDvid(1)
+ version.setDid(ownedDataset.getDid)
+ version.setName("v1")
+ version.setCreatorUid(owner.getUid)
+ version.setVersionHash("abc123")
+ version
+ }
+
+ // Another user's private dataset: the only fixture whose refusal comes from
read access
+ // rather than from the repository not existing.
+ private val strangersDataset: Dataset = {
+ val dataset = new Dataset
+ dataset.setDid(2)
+ dataset.setName("private")
+ dataset.setDescription("")
+ dataset.setRepositoryName("dataset-2")
+ dataset.setOwnerUid(stranger.getUid)
+ dataset.setIsPublic(false)
+ dataset
+ }
+
+ private val strangersVersion: DatasetVersion = {
+ val version = new DatasetVersion
+ version.setDvid(2)
+ version.setDid(strangersDataset.getDid)
+ version.setName("v1")
+ version.setCreatorUid(stranger.getUid)
+ version.setVersionHash("def456")
+ version
+ }
+
+ private val computingUnit: WorkflowComputingUnit = {
+ val unit = new WorkflowComputingUnit
+ unit.setCuid(7)
+ unit.setUid(owner.getUid)
+ unit.setName("test-cu")
+ unit.setType(WorkflowComputingUnitTypeEnum.kubernetes)
+ unit
+ }
+
+ /** Records what the resource asked the node mounter to do, without any
HTTP. */
+ /** Records what the resource asked the node mounter to do, without any
HTTP. */
+ private class RecordingMounter extends MounterClient("/nonexistent-token") {
+ val mounts: mutable.Buffer[(String, Int, String, String, String, String,
String)] =
+ mutable.Buffer()
+ var failWith: Option[Throwable] = None
+
+ override def mount(
+ nodeIp: String,
+ port: Int,
+ cuid: String,
+ repositoryName: String,
+ commitHash: String,
+ jwt: String,
+ fileServiceBase: String
+ ): String = {
+ failWith.foreach(throw _)
+ mounts += ((nodeIp, port, cuid, repositoryName, commitHash, jwt,
fileServiceBase))
+ s"/var/lib/texera-mounts/$cuid/$repositoryName/$commitHash"
+ }
+
+ }
+
+ private val scheduledOnNode = new ComputingUnitNodeLocator(_ => None) {
+ override def nodeIpOf(cuid: Int): Option[String] = Some("10.0.0.4")
+ }
+
+ private val notScheduled = new ComputingUnitNodeLocator(_ => None) {
+ override def nodeIpOf(cuid: Int): Option[String] = None
+ }
+
+ private def resource(
+ mounter: MounterClient,
+ mounterEnabled: Boolean = true,
+ nodeLocator: ComputingUnitNodeLocator = scheduledOnNode,
+ mounterPort: Option[Int] = Some(8100),
+ fileServiceUrl: Option[String] = Some("http://file-service-svc:9092";)
+ ) =
+ new ComputingUnitMountResource(
+ mounterEnabled,
+ mounterPort,
+ fileServiceUrl,
+ nodeLocator,
+ mounter
+ )
+
+ private def sessionOf(user: User) = new SessionUser(user)
+
+ override protected def beforeAll(): Unit = {
+ initializeDBAndReplaceDSLContext()
+ val userDao = new UserDao(getDSLContext.configuration())
+ userDao.insert(owner)
+ userDao.insert(reader)
+ userDao.insert(stranger)
+ new
WorkflowComputingUnitDao(getDSLContext.configuration()).insert(computingUnit)
+ val datasetDao = new DatasetDao(getDSLContext.configuration())
+ datasetDao.insert(ownedDataset)
+ datasetDao.insert(strangersDataset)
+ val versionDao = new DatasetVersionDao(getDSLContext.configuration())
+ versionDao.insert(ownedVersion)
+ versionDao.insert(strangersVersion)
+ val access = new ComputingUnitUserAccess()
+ access.setCuid(computingUnit.getCuid)
+ access.setUid(reader.getUid)
+ access.setPrivilege(PrivilegeEnum.READ)
+ new
ComputingUnitUserAccessDao(getDSLContext.configuration()).insert(access)
+ }
+
+ override protected def afterAll(): Unit = closeConnectionPool()
+
+ "mount" should "forward the request to the mounter on the unit's own node"
in {
+ val mounter = new RecordingMounter
+ val info = resource(mounter).mount(7, MountRequest("dataset-1", "abc123"),
sessionOf(owner))
+
+ info.repositoryName shouldBe "dataset-1"
+ info.commitHash shouldBe "abc123"
+ info.mountPath shouldBe "/var/lib/texera-mounts/7/dataset-1/abc123"
+
+ val (nodeIp, port, cuid, repository, commit, jwt, fileServiceBase) =
mounter.mounts.head
+ nodeIp shouldBe "10.0.0.4"
+ port shouldBe 8100
+ cuid shouldBe "7"
+ repository shouldBe "dataset-1"
+ commit shouldBe "abc123"
+ fileServiceBase shouldBe "http://file-service-svc:9092";
+ jwt should not be empty
+ }
+
+ // The only fixture with READ and not WRITE: tells "any access is enough"
apart from
+ // "write access is required".
+ it should "refuse a read-only sharee, who may use the unit but not change
what it sees" in {
+ val mounter = new RecordingMounter
+ a[ForbiddenException] should be thrownBy
+ resource(mounter).mount(7, MountRequest("dataset-1", "abc123"),
sessionOf(reader))
+ mounter.mounts shouldBe empty
+ }
+
+ it should "refuse a repository the user cannot read" in {
+ val mounter = new RecordingMounter
+ a[ForbiddenException] should be thrownBy
+ resource(mounter).mount(7, MountRequest("dataset-2", "def456"),
sessionOf(owner))
+ // The mounter authorizes nothing, so a refusal has to happen before it is
asked.
+ mounter.mounts shouldBe empty
+ }
+
+ it should "refuse a repository that does not exist" in {
+ val mounter = new RecordingMounter
+ a[ForbiddenException] should be thrownBy
+ resource(mounter).mount(7, MountRequest("dataset-404", "abc123"),
sessionOf(owner))
+ mounter.mounts shouldBe empty
+ }
+
+ it should "refuse a commit that belongs to another repository" in {
+ val mounter = new RecordingMounter
+ a[ForbiddenException] should be thrownBy
+ resource(mounter).mount(7, MountRequest("dataset-1", "def456"),
sessionOf(owner))
+ mounter.mounts shouldBe empty
+ }
+
+ it should "refuse a user with no access to the computing unit" in {
+ val mounter = new RecordingMounter
+ a[ForbiddenException] should be thrownBy
+ resource(mounter).mount(7, MountRequest("dataset-1", "abc123"),
sessionOf(stranger))
+ mounter.mounts shouldBe empty
+ }
+
+ it should "refuse a computing unit that does not exist" in {
+ val mounter = new RecordingMounter
+ a[ForbiddenException] should be thrownBy
+ resource(mounter).mount(999, MountRequest("dataset-1", "abc123"),
sessionOf(owner))
+ mounter.mounts shouldBe empty
+ }
+
+ it should "report a misconfigured port rather than dialling a wrong one" in {
+ val mounter = new RecordingMounter
+ an[InternalServerErrorException] should be thrownBy
+ resource(mounter, mounterPort = None)
+ .mount(7, MountRequest("dataset-1", "abc123"), sessionOf(owner))
+ mounter.mounts shouldBe empty
+ }
+
+ it should "report a missing file-service address rather than mounting
against nothing" in {
+ val mounter = new RecordingMounter
+ an[InternalServerErrorException] should be thrownBy
+ resource(mounter, fileServiceUrl = None)
+ .mount(7, MountRequest("dataset-1", "abc123"), sessionOf(owner))
+ mounter.mounts shouldBe empty
+ }
+
+ it should "answer plainly when the deployment did not enable mounting" in {
+ val mounter = new RecordingMounter
+ a[ServiceUnavailableException] should be thrownBy
+ resource(mounter, mounterEnabled = false)
+ .mount(7, MountRequest("dataset-1", "abc123"), sessionOf(owner))
+ mounter.mounts shouldBe empty
+ }
+
+ it should "refuse while the unit's pod is not on a node yet" in {
+ val mounter = new RecordingMounter
+ a[BadRequestException] should be thrownBy
+ resource(mounter, nodeLocator = notScheduled)
+ .mount(7, MountRequest("dataset-1", "abc123"), sessionOf(owner))
+ mounter.mounts shouldBe empty
+ }
+
+ it should "report a rejected path as a bad request rather than a server
error" in {
+ val mounter = new RecordingMounter
+ mounter.failWith = Some(new IllegalArgumentException("repositoryName must
be a single segment"))
+ val failure = the[BadRequestException] thrownBy
+ resource(mounter).mount(7, MountRequest("../evil", "abc123"),
sessionOf(owner))
+ failure.getMessage should include("repositoryName")
+ }
+
+ it should "relay a refusal from the mounter as a bad request" in {
+ val mounter = new RecordingMounter
+ mounter.failWith = Some(new MounterClient.MounterRequestException(400,
"mounter said no"))
+ a[BadRequestException] should be thrownBy
+ resource(mounter).mount(7, MountRequest("dataset-1", "abc123"),
sessionOf(owner))
+ }
+}
diff --git
a/access-control-service/src/test/scala/org/apache/texera/service/util/ComputingUnitNodeLocatorSpec.scala
b/access-control-service/src/test/scala/org/apache/texera/service/util/ComputingUnitNodeLocatorSpec.scala
new file mode 100644
index 0000000000..33f3a6a034
--- /dev/null
+++
b/access-control-service/src/test/scala/org/apache/texera/service/util/ComputingUnitNodeLocatorSpec.scala
@@ -0,0 +1,66 @@
+/*
+ * 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.util
+
+import com.fasterxml.jackson.databind.{JsonNode, ObjectMapper}
+import org.apache.texera.common.config.KubernetesConfig
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+import scala.collection.mutable
+
+class ComputingUnitNodeLocatorSpec extends AnyFlatSpec with Matchers {
+
+ private val mapper = new ObjectMapper()
+ private def pod(json: String): JsonNode = mapper.readTree(json)
+
+ private def locator(
+ pods: Map[String, JsonNode],
+ asked: mutable.Buffer[String] = mutable.Buffer()
+ ): ComputingUnitNodeLocator =
+ new ComputingUnitNodeLocator(name => { asked += name; pods.get(name) })
+
+ private val podName = s"${KubernetesConfig.computeUnitPodNamePrefix}-7"
+
+ "nodeIpOf" should "return the host IP of the computing unit's pod" in {
+ locator(Map(podName ->
pod("""{"status":{"hostIP":"10.0.0.4"}}"""))).nodeIpOf(7) shouldBe
+ Some("10.0.0.4")
+ }
+
+ it should "ask for the pod named by the configured prefix and the cuid" in {
+ val asked = mutable.Buffer[String]()
+ locator(Map.empty, asked).nodeIpOf(7)
+ asked should contain only podName
+ }
+
+ it should "return None when the computing unit has no pod" in {
+ locator(Map.empty).nodeIpOf(7) shouldBe None
+ }
+
+ it should "return None while the pod is not scheduled yet" in {
+ locator(Map(podName ->
pod("""{"status":{"phase":"Pending"}}"""))).nodeIpOf(7) shouldBe None
+ locator(Map(podName -> pod("""{"status":{"hostIP":""}}"""))).nodeIpOf(7)
shouldBe None
+ }
+
+ it should "propagate a lookup failure rather than reporting the unit as
unscheduled" in {
+ val failing = new ComputingUnitNodeLocator(_ => throw new
IllegalStateException("forbidden"))
+ an[IllegalStateException] should be thrownBy failing.nodeIpOf(7)
+ }
+}
diff --git
a/access-control-service/src/test/scala/org/apache/texera/service/util/MounterClientSpec.scala
b/access-control-service/src/test/scala/org/apache/texera/service/util/MounterClientSpec.scala
new file mode 100644
index 0000000000..beb14509ca
--- /dev/null
+++
b/access-control-service/src/test/scala/org/apache/texera/service/util/MounterClientSpec.scala
@@ -0,0 +1,153 @@
+/*
+ * 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.util
+
+import com.sun.net.httpserver.{HttpExchange, HttpServer}
+import org.scalatest.BeforeAndAfterAll
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+import java.net.InetSocketAddress
+import java.nio.charset.StandardCharsets
+import java.nio.file.Files
+import scala.collection.mutable
+
+class MounterClientSpec extends AnyFlatSpec with Matchers with
BeforeAndAfterAll {
+
+ private var server: HttpServer = _
+ private var port: Int = _
+ private var client: MounterClient = _
+
+ private val received = mutable.Map[String, (String, String, String)]()
+ private val authorization = mutable.Map[String, String]()
+ private var replyWith: Option[(Int, String)] = None
+
+ private def bodyOf(exchange: HttpExchange): String =
+ new String(exchange.getRequestBody.readAllBytes(), StandardCharsets.UTF_8)
+
+ private def reply(exchange: HttpExchange, status: Int, body: String): Unit =
{
+ val bytes = body.getBytes(StandardCharsets.UTF_8)
+ exchange.getResponseHeaders.add("Content-Type", "application/json")
+ exchange.sendResponseHeaders(status, bytes.length.toLong)
+ exchange.getResponseBody.write(bytes)
+ exchange.close()
+ }
+
+ private def record(exchange: HttpExchange, path: String): Unit = {
+ authorization(path) =
Option(exchange.getRequestHeaders.getFirst("Authorization")).getOrElse("")
+ received(path) = (
+ exchange.getRequestMethod,
+ Option(exchange.getRequestURI.getQuery).getOrElse(""),
+ bodyOf(exchange)
+ )
+ }
+
+ override def beforeAll(): Unit = {
+ val tokenFile = Files.createTempFile("mounter-token", "")
+ Files.writeString(tokenFile, "the-service-account-token\n")
+ tokenFile.toFile.deleteOnExit()
+ client = new MounterClient(tokenFile.toString)
+
+ server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0)
+ server.createContext(
+ "/mount",
+ (exchange: HttpExchange) => {
+ record(exchange, "/mount")
+ replyWith match {
+ case Some((status, body)) => reply(exchange, status, body)
+ case None =>
+ reply(exchange, 200,
"""{"mountPath":"/var/lib/texera-mounts/7/dataset-1/abc123"}""")
+ }
+ }
+ )
+ server.start()
+ port = server.getAddress.getPort
+ }
+
+ override def afterAll(): Unit = if (server != null) server.stop(0)
+
+ private val nodeIp = "127.0.0.1"
+
+ "MounterClient.mount" should "post the mount request and return the
mounter's path" in {
+ val path = client.mount(nodeIp, port, "7", "dataset-1", "abc123",
"user-jwt", "http://fs:9092";)
+
+ path shouldBe "/var/lib/texera-mounts/7/dataset-1/abc123"
+ val (method, _, requestBody) = received("/mount")
+ method shouldBe "POST"
+ requestBody should include(""""cuid":"7"""")
+ requestBody should include(""""repositoryName":"dataset-1"""")
+ requestBody should include(""""commitHash":"abc123"""")
+ requestBody should include(""""jwt":"user-jwt"""")
+ requestBody should include(""""fileServiceBase":"http://fs:9092"""";)
+ }
+
+ it should "identify itself with the projected service-account token" in {
+ client.mount(nodeIp, port, "7", "dataset-1", "abc123", "user-jwt",
"http://fs:9092";)
+ authorization("/mount") shouldBe "Bearer the-service-account-token"
+ }
+
+ it should "carry the mounter's status back on a refusal" in {
+ replyWith = Some((400, """{"error":"nope"}"""))
+ try {
+ val failure = the[MounterClient.MounterRequestException] thrownBy
+ client.mount(nodeIp, port, "7", "dataset-1", "abc123", "user-jwt",
"http://fs:9092";)
+ failure.status shouldBe 400
+ } finally replyWith = None
+ }
+
+ it should "fail rather than report a mount when the mounter's success names
no path" in {
+ Seq("{}", """{"mountPath":null}""", """{"mountPath":""}""").foreach { body
=>
+ replyWith = Some((200, body))
+ try {
+ an[IllegalStateException] should be thrownBy
+ client.mount(nodeIp, port, "7", "dataset-1", "abc123", "user-jwt",
"http://fs:9092";)
+ } finally replyWith = None
+ }
+ }
+
+ // The escapes reported on the infrastructure PR: each would otherwise be
joined into the
+ // mount path, and the directory created before the mounter's own validation
could matter.
+ it should "refuse a cuid that is not a single numeric segment, without
calling the mounter" in {
+ received.remove("/mount")
+ Seq("5/../8", "../..", "/absolute", "", "7x").foreach { cuid =>
+ an[IllegalArgumentException] should be thrownBy
+ client.mount(nodeIp, port, cuid, "dataset-1", "abc123", "jwt",
"http://fs:9092";)
+ }
+ received should not contain key("/mount")
+ }
+
+ it should "refuse a repository or commit that is not a single safe segment"
in {
+ received.remove("/mount")
+ Seq("../evil", "a/b", "-o", "", ".hidden/../x").foreach { bad =>
+ an[IllegalArgumentException] should be thrownBy
+ client.mount(nodeIp, port, "7", bad, "abc123", "jwt", "http://fs:9092";)
+ an[IllegalArgumentException] should be thrownBy
+ client.mount(nodeIp, port, "7", "dataset-1", bad, "jwt",
"http://fs:9092";)
+ }
+ received should not contain key("/mount")
+ }
+
+ it should "fail loudly when the service-account token is missing" in {
+ val withoutToken = new MounterClient("/nonexistent/mounter/token")
+ val failure = the[IllegalStateException] thrownBy
+ withoutToken.mount(nodeIp, port, "7", "dataset-1", "abc123", "jwt",
"http://fs:9092";)
+ failure.getMessage should include("/nonexistent/mounter/token")
+ }
+}
diff --git
a/bin/k8s/templates/base/access-control-service/access-control-service-deployment.yaml
b/bin/k8s/templates/base/access-control-service/access-control-service-deployment.yaml
index 62bb8f3b21..7d63720365 100644
---
a/bin/k8s/templates/base/access-control-service/access-control-service-deployment.yaml
+++
b/bin/k8s/templates/base/access-control-service/access-control-service-deployment.yaml
@@ -60,6 +60,18 @@ spec:
- name: KUBERNETES_JUPYTER_BASE_URL
value: {{ include "texera.jupyter.basePath" . }}
{{- end }}
+ {{- if .Values.mounter.enabled }}
+ # Passed from the chart rather than left to the kubernetes.conf
defaults, so that
+ # changing a value here reaches the service that reads it.
+ - name: KUBERNETES_MOUNTER_ENABLED
+ value: "{{ .Values.mounter.enabled }}"
+ - name: KUBERNETES_COMPUTE_UNIT_POD_NAME_PREFIX
+ value: {{ .Values.workflowComputingUnitPool.podNamePrefix }}
+ - name: KUBERNETES_MOUNTER_PORT
+ value: "{{ .Values.mounter.port }}"
+ - name: FILE_SERVICE_URL
+ value: http://{{ .Values.fileService.name }}-svc:{{
.Values.fileService.service.port }}
+ {{- end }}
{{- if .Values.litellm.enabled }}
# LLM gateway used to serve /api/chat and /api/models to the agent
service.
- name: LITELLM_BASE_URL
diff --git
a/bin/k8s/templates/base/access-control-service/access-control-service-service-account.yaml
b/bin/k8s/templates/base/access-control-service/access-control-service-service-account.yaml
index 44cd6bbf07..c4e603e799 100644
---
a/bin/k8s/templates/base/access-control-service/access-control-service-service-account.yaml
+++
b/bin/k8s/templates/base/access-control-service/access-control-service-service-account.yaml
@@ -18,16 +18,13 @@
# Dedicated identity for the access-control-service.
#
-# The access-control-service is intended to become the only component allowed
to ask the
-# per-node mounter to mount a dataset: it is already the JWT and
computing-unit-access
-# authorization proxy, so it is the natural place for the decision "may this
user mount
-# onto this CU?". Giving it its own identity now is what makes that switch a
config
-# change later, rather than a redesign -- running as the namespace's `default`
-# ServiceAccount (shared with every pod that does not name one) would make the
mounter
-# unable to tell this service apart from anything else.
+# The access-control-service is the only component allowed to ask the per-node
mounter to
+# mount a repository: it is already the JWT and computing-unit-access
authorization proxy,
+# so it is the natural place for the decision "may this user mount onto this
CU?" --
+# running as the namespace's `default` ServiceAccount (shared with every pod
that does not
+# name one) would make the mounter unable to tell this service apart from
anything else.
#
-# The enforcement mechanism already exists and is live in this PR; only the
identity in
-# the allow-list is still provisional:
+# How that is enforced:
#
# 1. The calling pod mounts a projected `serviceAccountToken` volume bound
to the
# audience `texera-mounter` and sends that token as a Bearer header on
each mounter
@@ -44,22 +41,41 @@
# 3. The mounter's own ServiceAccount is bound to the built-in
`system:auth-delegator`
# ClusterRole, which is what grants it permission to create TokenReviews.
#
-# TODO(dataset-mount): today `mounter.allowedCallers` defaults to the
computing-unit
-# manager, because that is the service which actually calls the mounter. Point
it at this
-# account -- and move the mount endpoints behind this service -- once
access-control-service
-# takes over as the mount authority. Nothing else has to change.
-#
-# Either way, computing-unit pods are never an accepted caller even though
they can reach
+# Computing-unit pods are never an accepted caller even though they can reach
# the mounter's hostPort: they hold no token for this audience, so a mount
request forged
# from user code fails the TokenReview regardless of what it puts in the
request body. And
# because authenticating the caller only establishes who is asking, the
mounter still
# validates every path component of the request itself.
#
-# This account needs no RBAC rules: it is an identity to authenticate as, not
a client
-# of the Kubernetes API.
+# The Role below lets this service read which node a computing unit's pod runs
on, so it can
+# forward a mount to that node's mounter.
apiVersion: v1
kind: ServiceAccount
metadata:
name: {{ .Values.accessControlService.serviceAccountName }}
namespace: {{ .Release.Namespace }}
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: Role
+metadata:
+ name: {{ .Release.Name }}-access-control-service-pod-reader
+ namespace: {{ .Values.workflowComputingUnitPool.namespace }}
+rules:
+ - apiGroups: [""]
+ resources: ["pods"]
+ verbs: ["get"]
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: RoleBinding
+metadata:
+ name: {{ .Release.Name }}-access-control-service-pod-reader-binding
+ namespace: {{ .Values.workflowComputingUnitPool.namespace }}
+subjects:
+ - kind: ServiceAccount
+ name: {{ .Values.accessControlService.serviceAccountName }}
+ namespace: {{ .Release.Namespace }}
+roleRef:
+ kind: Role
+ name: {{ .Release.Name }}-access-control-service-pod-reader
+ apiGroup: rbac.authorization.k8s.io
{{- end }}
diff --git a/build.sbt b/build.sbt
index 3853666c68..cb16bda952 100644
--- a/build.sbt
+++ b/build.sbt
@@ -122,7 +122,7 @@ ThisBuild / excludeDependencies += ExclusionRule("log4j",
"log4j")
lazy val Util = (project in file("common/util")).settings(commonModuleSettings)
lazy val DAO = (project in file("common/dao")).settings(commonModuleSettings)
lazy val Config = (project in
file("common/config")).settings(commonModuleSettings)
-lazy val Resource = (project in
file("common/resource")).settings(commonModuleSettings)
+lazy val Resource = (project in
file("common/resource")).settings(commonModuleSettings).dependsOn(DAO)
lazy val Auth = (project in file("common/auth"))
.settings(commonModuleSettings)
.configs(Test)
diff --git
a/file-service/src/main/scala/org/apache/texera/service/resource/ResourceAccess.scala
b/common/resource/src/main/scala/org/apache/texera/service/resource/ResourceAccess.scala
similarity index 100%
rename from
file-service/src/main/scala/org/apache/texera/service/resource/ResourceAccess.scala
rename to
common/resource/src/main/scala/org/apache/texera/service/resource/ResourceAccess.scala
diff --git
a/file-service/src/main/scala/org/apache/texera/service/resource/ResourceTables.scala
b/common/resource/src/main/scala/org/apache/texera/service/resource/ResourceTables.scala
similarity index 100%
rename from
file-service/src/main/scala/org/apache/texera/service/resource/ResourceTables.scala
rename to
common/resource/src/main/scala/org/apache/texera/service/resource/ResourceTables.scala
