feat(agent-service): authenticate requests via JWT and gateway ext_authz

access-control-service/src/main/scala/org/apache/texera/service/resource/AccessControlResource.scala

@@ -23,13 +23,13 @@ import com.typesafe.scalalogging.LazyLogging
 import jakarta.annotation.security.{PermitAll, RolesAllowed}
 import jakarta.ws.rs.client.{Client, ClientBuilder, Entity}
 import jakarta.ws.rs.core._
-import jakarta.ws.rs.{Consumes, DELETE, GET, POST, PUT, Path, Produces}
+import jakarta.ws.rs.{Consumes, DELETE, GET, PATCH, POST, PUT, Path, Produces}
 import org.apache.texera.auth.JwtParser.parseToken
 import org.apache.texera.auth.SessionUser
 import org.apache.texera.auth.util.{ComputingUnitAccess, HeaderField}
 import org.apache.texera.common.config.{GuiConfig, KubernetesConfig, LLMConfig}
 import org.apache.texera.dao.SqlServer
-import org.apache.texera.dao.jooq.generated.enums.PrivilegeEnum
+import org.apache.texera.dao.jooq.generated.enums.{PrivilegeEnum, UserRoleEnum}
 import org.apache.texera.dao.jooq.generated.tables.daos.{UserJupyterDao, WorkflowComputingUnitDao}
 
 import java.net.{URI, URLDecoder}
@@ -47,6 +47,8 @@ object AccessControlResource extends LazyLogging {
   private val apiExecutionsStats: Regex = """.*/api/executions/[0-9]+/stats/[0-9]+.*""".r
   private val apiExecutionsResultExport: Regex = """.*/api/executions/result/export.*""".r
   private val pveRoute: Regex = """^/?(?:auth/)?(?:api/|wsapi/)?pve(?:/.*)?$""".r
+  // Agent service: authenticate any /api/agents request (Phase 1 — see #5561).
+  private val apiAgents: Regex = """^/?(?:auth/)?api/agents(?:/.*)?$""".r
   // Path patterns whose cuid lives in the URL path rather than the query string.
   private val pvePvesCuidPath: Regex = """^/?(?:auth/)?(?:api/|wsapi/)?pve/pves/([0-9]+)$""".r
   private val pvePackagesCuidPath: Regex =
@@ -84,6 +86,8 @@ object AccessControlResource extends LazyLogging {
       case wsapiWorkflowWebsocket() | apiExecutionsStats() | apiExecutionsResultExport() |
           pveRoute() =>
         checkComputingUnitAccess(uriInfo, headers, bodyOpt)
+      case apiAgents() =>
+        checkAgentAccess(uriInfo, headers, bodyOpt)
       case _ =>
         logger.warn(s"No authorization logic for path: $path. Denying access.")
         Response.status(Response.Status.FORBIDDEN).build()
@@ -125,6 +129,60 @@ object AccessControlResource extends LazyLogging {
     }
   }
 
+  // Extract the bearer token from the access-token query param, the
+  // Authorization header, or a "token" field in the body (in that order).
+  private def extractBearerToken(
+      uriInfo: UriInfo,
+      headers: HttpHeaders,
+      bodyOpt: Option[String]
+  ): String = {
+    val qToken = Option(uriInfo.getQueryParameters().getFirst("access-token"))
+      .map(_.trim)
+      .filter(_.nonEmpty)
+    val hToken = Option(headers.getRequestHeader("Authorization"))
+      .flatMap(_.asScala.headOption)
+      .map(_.replaceFirst("(?i)^Bearer\\s+", "")) // case-insensitive "Bearer "
+      .map(_.trim)
+      .filter(_.nonEmpty)
+    val bToken = bodyOpt.flatMap(extractTokenFromBody)
+    qToken.orElse(hToken).orElse(bToken).getOrElse("")
+  }
+
+  // Phase 1 agent authorization: authenticate the JWT and require a
+  // REGULAR/ADMIN role. Any such user may reach any agent for now; per-agent
+  // ownership is deferred (see #5302 / #5561). On success, return the user
+  // identity headers for future per-agent authorization in agent-service.
+  private def checkAgentAccess(
+      uriInfo: UriInfo,
+      headers: HttpHeaders,
+      bodyOpt: Option[String]
+  ): Response = {
+    val token = extractBearerToken(uriInfo, headers, bodyOpt)
+    val userSession: Optional[SessionUser] =
+      try parseToken(token)
+      catch {
+        case e: Exception =>
+          logger.error(s"Failed parsing token for agent request: $e")
+          Optional.empty()
+      }
+
+    if (userSession.isEmpty) {
+      return Response.status(Response.Status.UNAUTHORIZED).build()
+    }
+
+    val user = userSession.get()
+    if (!(user.isRoleOf(UserRoleEnum.REGULAR) || user.isRoleOf(UserRoleEnum.ADMIN))) {
+      return Response.status(Response.Status.FORBIDDEN).build()
+    }
+
+    Response
+      .ok()
+      .header(HeaderField.UserId, user.getUid.toString)
+      .header(HeaderField.UserName, user.getName)
+      .header(HeaderField.UserEmail, user.getEmail)
+      .build()
+  }
+
   private def checkComputingUnitAccess(
       uriInfo: UriInfo,
       headers: HttpHeaders,
@@ -141,16 +199,7 @@ object AccessControlResource extends LazyLogging {
       s"Request URI: ${uriInfo.getRequestUri} and headers: ${headers.getRequestHeaders.asScala} and queryParams: $queryParams"
     )
 
-    val token: String = {
-      val qToken = queryParams.get("access-token").filter(_.nonEmpty)
-      val hToken = Option(headers.getRequestHeader("Authorization"))
-        .flatMap(_.asScala.headOption)
-        .map(_.replaceFirst("(?i)^Bearer\\s+", "")) // case-insensitive "Bearer "
-        .map(_.trim)
-        .filter(_.nonEmpty)
-      val bToken = bodyOpt.flatMap(extractTokenFromBody)
-      qToken.orElse(hToken).orElse(bToken).getOrElse("")
-    }
+    val token: String = extractBearerToken(uriInfo, headers, bodyOpt)
     logger.info(s"token extracted from request $token")
 
     val cuid = queryParams.get("cuid").filter(_.nonEmpty).getOrElse {
@@ -306,6 +355,16 @@ class AccessControlResource extends LazyLogging {
     AccessControlResource.authorize(uriInfo, headers, Option(body).map(_.trim).filter(_.nonEmpty))
   }
 
+  @PATCH
+  @Path("/{path:.*}")
+  def authorizePatch(
+      @Context uriInfo: UriInfo,
+      @Context headers: HttpHeaders,
+      body: String
+  ): Response = {
+    AccessControlResource.authorize(uriInfo, headers, Option(body).map(_.trim).filter(_.nonEmpty))
+  }
+
   @DELETE
   @Path("/{path:.*}")
   def authorizeDelete(

access-control-service/src/test/scala/org/apache/texera/service/resource/AgentAccessAuthSpec.scala

@@ -0,0 +1,112 @@
+/*
+ * 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.core.{HttpHeaders, MultivaluedHashMap, Response, UriInfo}
+import org.apache.texera.auth.JwtAuth
+import org.apache.texera.dao.jooq.generated.enums.UserRoleEnum
+import org.apache.texera.dao.jooq.generated.tables.pojos.User
+import org.mockito.Mockito.{mock, when}
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+import java.net.URI
+import java.util.Collections
+
+/**
+  * Exercises the `/api/agents` branch of [[AccessControlResource.authorize]] —
+  * the Phase 1 ext_authz gate that authenticates the JWT and requires a
+  * REGULAR/ADMIN role. Tokens are minted with the same [[JwtAuth]] the rest of
+  * Texera uses, so this is a real signature round-trip against auth.conf.
+  */
+class AgentAccessAuthSpec extends AnyFlatSpec with Matchers {
+
+  private def token(role: UserRoleEnum): String = {
+    val u = new User()
+    u.setUid(7)
+    u.setName("agent-user")
+    u.setEmail("agent-user@example.com")
+    u.setRole(role)
+    JwtAuth.jwtToken(JwtAuth.jwtClaims(u))
+  }
+
+  private def buildMocks(path: String, authHeader: Option[String]): (UriInfo, HttpHeaders) = {
+    val uriInfo = mock(classOf[UriInfo])
+    when(uriInfo.getPath).thenReturn(path)
+    when(uriInfo.getRequestUri).thenReturn(URI.create(s"http://localhost/$path"))
+    when(uriInfo.getQueryParameters()).thenReturn(new MultivaluedHashMap[String, String]())
+
+    val headers = mock(classOf[HttpHeaders])
+    when(headers.getRequestHeaders).thenReturn(new MultivaluedHashMap[String, String]())
+    when(headers.getRequestHeader("Authorization")).thenReturn(
+      authHeader.map((h: String) => Collections.singletonList(h)).orNull
+    )
+
+    (uriInfo, headers)
+  }
+
+  private def authorize(path: String, authHeader: Option[String]): Response = {
+    val (uriInfo, headers) = buildMocks(path, authHeader)
+    AccessControlResource.authorize(uriInfo, headers)
+  }
+
+  private val agentPath = "auth/api/agents/agent-1"
+
+  "authorize on /api/agents" should "return 200 for a REGULAR user" in {
+    authorize(agentPath, Some(s"Bearer ${token(UserRoleEnum.REGULAR)}")).getStatus shouldBe 200
+  }
+
+  it should "return 200 for an ADMIN user" in {
+    authorize(agentPath, Some(s"Bearer ${token(UserRoleEnum.ADMIN)}")).getStatus shouldBe 200
+  }
+
+  it should "forward the user identity headers on success" in {
+    val resp = authorize(agentPath, Some(s"Bearer ${token(UserRoleEnum.REGULAR)}"))
+    resp.getHeaderString("x-user-id") shouldBe "7"
+    resp.getHeaderString("x-user-name") shouldBe "agent-user"
+    resp.getHeaderString("x-user-email") shouldBe "agent-user@example.com"
+  }
+
+  it should "return 401 when no token is present" in {
+    authorize(agentPath, None).getStatus shouldBe 401
+  }
+
+  it should "return 401 for a malformed / mis-signed token" in {
+    authorize(agentPath, Some("Bearer not-a-real-jwt")).getStatus shouldBe 401
+  }
+
+  it should "return 403 for an INACTIVE-role user" in {
+    authorize(agentPath, Some(s"Bearer ${token(UserRoleEnum.INACTIVE)}")).getStatus shouldBe 403
+  }
+
+  it should "authorize PATCH requests through authorizePatch handler" in {
+    val (uriInfo, headers) =
+      buildMocks(s"$agentPath/settings", Some(s"Bearer ${token(UserRoleEnum.REGULAR)}"))
+    val resource = new AccessControlResource()
+    resource.authorizePatch(uriInfo, headers, """{"model": "test"}""").getStatus shouldBe 200
+  }
+
+  it should "not match dynamic wsapi routes containing api/agents and deny access" in {
+    authorize(
+      "auth/wsapi/x/api/agents",
+      Some(s"Bearer ${token(UserRoleEnum.REGULAR)}")
+    ).getStatus shouldBe 403
+  }
+}

agent-service/src/api/auth-api.spec.ts

@@ -18,23 +18,35 @@
  */
 
 import { describe, expect, test } from "bun:test";
+import { createHmac } from "node:crypto";
+import { JWT_SECRET } from "../config/jwt";
 import { createAuthHeaders, extractBearerToken, extractUserFromToken, validateToken } from "./auth-api";
 
-/**
- * Builds a well-formed three-segment token carrying the given payload. All three segments are
- * required — `decodeJWT` rejects anything that is not exactly three — but only the payload one is
- * decoded, so the header and signature just have to be present.
- */
-function tokenWith(payload: Record<string, unknown>): string {
-  const encoded = Buffer.from(JSON.stringify(payload), "utf-8").toString("base64");
-  return `header.${encoded}.signature`;
+function b64url(input: string | Buffer): string {
+  return Buffer.from(input).toString("base64url");
 }
 
 /** `exp` is a UNIX second count, so expiries are built in seconds throughout. */
 function nowInSeconds(): number {
   return Math.floor(Date.now() / 1000);
 }
 
+/**
+ * Builds a properly signed HS256 JWT using JWT_SECRET with default valid fields.
+ */
+function tokenWith(payload: Record<string, unknown>, secret = JWT_SECRET): string {
+  const fullPayload = {
+    sub: "test-user",
+    userId: 1,
+    exp: nowInSeconds() + 3600,
+    ...payload,
+  };
+  const header = b64url(JSON.stringify({ alg: "HS256", typ: "JWT" }));
+  const body = b64url(JSON.stringify(fullPayload));
+  const signature = b64url(createHmac("sha256", new TextEncoder().encode(secret)).update(`${header}.${body}`).digest());
+  return `${header}.${body}.${signature}`;
+}
+
 describe("extractUserFromToken", () => {
   test("maps the JWT payload onto the user record", () => {
     const token = tokenWith({ userId: 42, sub: "ada", email: "ada@example.com", role: "ADMIN" });
@@ -48,30 +60,31 @@ describe("extractUserFromToken", () => {
   });
 
   test("defaults a missing email to empty and a missing role to REGULAR", () => {
-    // Role defaulting is a privilege decision: absent must mean least privilege, never admin.
-    const user = extractUserFromToken(tokenWith({ userId: 7, sub: "grace" }));
+    const token = tokenWith({ userId: 7, sub: "grace", email: undefined, role: undefined });
+    const user = extractUserFromToken(token);
 
     expect(user.email).toBe("");
     expect(user.role).toBe("REGULAR");
   });
 
-  test("rejects a token missing its signature segment even when the payload decodes", () => {
-    // The payload here is perfectly good JSON, so only the segment-count check stands between an
-    // unsigned token and a populated user record.
-    const payload = Buffer.from(JSON.stringify({ userId: 1, sub: "ada" }), "utf-8").toString("base64");
-
-    expect(() => extractUserFromToken(`header.${payload}`)).toThrow(/Failed to decode JWT/);
+  test("rejects a token missing its signature segment", () => {
+    const payload = Buffer.from(JSON.stringify({ userId: 1, sub: "ada" }), "utf-8").toString("base64url");
+    expect(() => extractUserFromToken(`header.${payload}`)).toThrow(/Invalid or expired token/);
   });
 
   test("rejects a token with too many segments", () => {
-    const payload = Buffer.from(JSON.stringify({ userId: 1, sub: "ada" }), "utf-8").toString("base64");
-
-    expect(() => extractUserFromToken(`header.${payload}.signature.extra`)).toThrow(/Failed to decode JWT/);
+    const payload = Buffer.from(JSON.stringify({ userId: 1, sub: "ada" }), "utf-8").toString("base64url");
+    expect(() => extractUserFromToken(`header.${payload}.signature.extra`)).toThrow(/Invalid or expired token/);
   });
 
   test("rejects a token whose payload is not JSON", () => {
-    const notJson = Buffer.from("<html>", "utf-8").toString("base64");
-    expect(() => extractUserFromToken(`header.${notJson}.signature`)).toThrow(/Failed to decode JWT/);
+    const notJson = Buffer.from("<html>", "utf-8").toString("base64url");
+    expect(() => extractUserFromToken(`header.${notJson}.signature`)).toThrow(/Invalid or expired token/);
+  });
+
+  test("rejects a mis-signed token", () => {
+    const badToken = tokenWith({ userId: 1, sub: "ada" }, "wrong-secret");
+    expect(() => extractUserFromToken(badToken)).toThrow(/Invalid or expired token/);
   });
 });
 
@@ -84,21 +97,20 @@ describe("validateToken", () => {
     expect(validateToken(tokenWith({ userId: 1, exp: nowInSeconds() - 60 }))).toBe(false);
   });
 
-  test("treats a token with no expiry as valid forever", () => {
-    // Deliberate: tokens minted without `exp` never expire. Pinned because it is a policy choice
-    // that reads like an oversight, and flipping it would lock out every non-expiring token.
-    expect(validateToken(tokenWith({ userId: 1 }))).toBe(true);
+  test("rejects a token without exp (required claim)", () => {
+    const header = b64url(JSON.stringify({ alg: "HS256", typ: "JWT" }));
+    const body = b64url(JSON.stringify({ sub: "test-user", userId: 1 }));
+    const signature = b64url(
+      createHmac("sha256", new TextEncoder().encode(JWT_SECRET)).update(`${header}.${body}`).digest()
+    );
+    expect(validateToken(`${header}.${body}.${signature}`)).toBe(false);
   });
 
-  test("rejects a malformed token rather than letting the decode failure escape", () => {
-    // isTokenExpired swallows the decode error and reports "expired", so callers get false here
-    // instead of an exception. A caller that only catches would otherwise let a junk token through.
+  test("rejects a malformed token rather than letting error escape", () => {
     expect(validateToken("not-a-token")).toBe(false);
   });
 
   test("expiry is compared in seconds, not milliseconds", () => {
-    // exp is a UNIX second count; reading it as milliseconds would place every real token in 1970
-    // and reject it. A far-future second count must still validate.
     expect(validateToken(tokenWith({ exp: nowInSeconds() + 3600 }))).toBe(true);
   });
 });