aicam commented on code in PR #6866: URL: https://github.com/apache/texera/pull/6866#discussion_r3929226028
########## file-service/src/main/scala/org/apache/texera/service/util/S3ProxyServlet.scala: ########## @@ -0,0 +1,260 @@ +/* + * 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.typesafe.scalalogging.LazyLogging +import jakarta.servlet.http.{HttpServlet, HttpServletRequest, HttpServletResponse} +import org.apache.texera.auth.JwtParser +import org.apache.texera.common.config.StorageConfig +import org.apache.texera.dao.SqlServer +import org.apache.texera.dao.SqlServer.withTransaction +import org.apache.texera.dao.jooq.generated.tables.daos.DatasetDao +import org.apache.texera.service.resource.DatasetAccessResource.userHasReadAccess +import software.amazon.awssdk.auth.credentials.AwsBasicCredentials +import software.amazon.awssdk.auth.signer.AwsS3V4Signer +import software.amazon.awssdk.auth.signer.params.AwsS3V4SignerParams +import software.amazon.awssdk.http.{SdkHttpFullRequest, SdkHttpMethod} +import software.amazon.awssdk.regions.Region + +import java.net.{URI, URLDecoder} +import java.net.http.{HttpClient, HttpRequest, HttpResponse} +import java.time.Duration +import java.util.concurrent.ConcurrentHashMap +import scala.jdk.CollectionConverters._ +import scala.jdk.OptionConverters._ + +/** + * Read-only, JWT-authenticated, re-signing reverse proxy in front of the LakeFS S3 + * gateway. A computing-unit pod's GeeseFS mount talks to this servlet using the pod's + * own per-user JWT as the S3 credential: the JWT is passed to GeeseFS as + * `AWS_ACCESS_KEY_ID`, so it rides in the request's SigV4/SigV2 `Authorization` header. + * Reusing the JWT that is already present in the pod means no separate mount credential + * is ever issued, stored, or made multi-replica-consistent. The servlet: + * + * 1. reads the JWT back out of the incoming `Authorization` header (the JWT is the + * bearer capability; the pod-side S3 signature is not re-validated, and no LakeFS + * credentials ever leave this service), + * 2. verifies the JWT and checks that its user has read access to the requested + * repository (the S3 bucket), using the same `userHasReadAccess` gate as the + * dataset REST endpoints, and + * 3. re-signs the request with the global LakeFS credentials and forwards it to the + * LakeFS S3 gateway, streaming the response back. + * + * Because requests are forwarded verbatim, the proxy behaves identically to a direct + * GeeseFS -> LakeFS-gateway mount, just re-authenticated. GeeseFS mounts read-only, so + * only GET and HEAD are handled. + */ +class S3ProxyServlet extends HttpServlet with LazyLogging { + + // The LakeFS S3 gateway shares the LakeFS server address: the configured API endpoint + // with the trailing /api/v1 suffix removed. + private val gatewayEndpoint: URI = + URI.create(StorageConfig.lakefsEndpoint.stripSuffix("/").stripSuffix("/api/v1")) + + private val lakefsCredentials = + AwsBasicCredentials.create(StorageConfig.lakefsUsername, StorageConfig.lakefsPassword) + + // The S3-specific SigV4 signer adds and signs the x-amz-content-sha256 header (which + // S3 / the LakeFS gateway require) and disables path double-encoding — both essential + // for the signature to validate. The generic Aws4Signer omits x-amz-content-sha256. + private val signer = AwsS3V4Signer.create() + + private val httpClient: HttpClient = HttpClient + .newBuilder() + .followRedirects(HttpClient.Redirect.NEVER) + .connectTimeout(Duration.ofSeconds(10)) + .build() + + private val forwardedResponseHeaderPrefixes = + Seq("content-", "etag", "last-modified", "accept-ranges", "x-amz-") + + // Short-lived cache of (uid, repository) -> read-access decision. A mount issues many + // range reads for the same repository, so this avoids a DB round-trip per request. It + // is a pure optimization: each replica caches independently and a miss just re-queries, + // so unlike a shared session store it needs no cross-replica consistency. + private val AuthCacheTtlMs = 60000L + private case class CachedDecision(allowed: Boolean, expiresAtMs: Long) + private val authCache = new ConcurrentHashMap[String, CachedDecision]() + + override def doGet(req: HttpServletRequest, resp: HttpServletResponse): Unit = + proxy(req, resp, SdkHttpMethod.GET, streamBody = true) + + override def doHead(req: HttpServletRequest, resp: HttpServletResponse): Unit = + proxy(req, resp, SdkHttpMethod.HEAD, streamBody = false) + + private def proxy( + req: HttpServletRequest, + resp: HttpServletResponse, + method: SdkHttpMethod, + streamBody: Boolean + ): Unit = { + val user = S3ProxyServlet + .extractCredentialToken(req.getHeader("Authorization")) + .flatMap(token => JwtParser.parseToken(token).toScala) + if (user.isEmpty) { + // GeeseFS probes the bucket unauthenticated on mount, so this is expected noise. + resp.sendError(HttpServletResponse.SC_FORBIDDEN, "missing or invalid user token") + return + } + + val uid = user.get.getUid + val repositoryName = S3ProxyServlet.bucketFromUri(req.getRequestURI) + if (repositoryName.isEmpty || !authorizedToRead(uid, repositoryName)) { + logger.warn( + s"user $uid denied mount access to repository '$repositoryName' for ${req.getRequestURI}" + ) + resp.sendError(HttpServletResponse.SC_FORBIDDEN, "no read access to the requested repository") + return + } + + try { + writeResponse(forward(req, method), resp, streamBody) + } catch { + case e: Exception => + logger.error(s"error proxying ${req.getRequestURI} to LakeFS gateway", e) + resp.sendError(HttpServletResponse.SC_BAD_GATEWAY, "upstream error") + } + } + + /** + * True iff `uid` has read access to the dataset backing `repositoryName`, cached for a + * short window. Read access to a repository grants read to all of its commits, so no + * per-commit check is needed: a session addresses a single repository's data and any + * version the user may already read. + */ + private def authorizedToRead(uid: Integer, repositoryName: String): Boolean = { + val now = System.currentTimeMillis() + val cacheKey = s"$uid:$repositoryName" + val cached = authCache.get(cacheKey) + if (cached != null && cached.expiresAtMs > now) { + return cached.allowed + } + + val allowed = withTransaction(SqlServer.getInstance().createDSLContext()) { ctx => + val datasets = new DatasetDao(ctx.configuration()) + .fetchByRepositoryName(repositoryName) + .asScala + .toList + datasets.nonEmpty && userHasReadAccess(ctx, datasets.head.getDid, uid) Review Comment: Fixed, and you were right that it is not exploitable — worth recording why, since it is stronger than the naming convention: LakeFS refuses a second repository with the same name, and dataset creation checked the name globally (`fetchByName`, not per-owner) before that. So one repository name is one row. The schema still does not say so, and the fix is free, so the lookup now requires exactly one match across `dataset` and `model` and denies (with a log) otherwise, rather than letting `.head` pick. One thing your comment led me to that does matter: I first tried parsing the `<type>-<id>` out of the name and looking it up by primary key. That is wrong — `sql/updates/15.sql` backfilled `repository_name` from the dataset's plain `name`, so upgraded deployments still have repositories called e.g. `my-data`, and parsing would have denied every one of them. Matching on the column handles both. (34c8a12d3) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
