feat(access-control-service): add access-control-service to authorize the requests to /wsapi and Computing Unit endpoints

core/amber/src/main/scala/edu/uci/ics/texera/web/ServletAwareConfigurator.scala

@@ -21,6 +21,9 @@ package edu.uci.ics.texera.web
 
 import com.typesafe.scalalogging.LazyLogging
 import edu.uci.ics.texera.auth.JwtAuth.jwtConsumer
+import edu.uci.ics.texera.auth.util.HeaderField
+import edu.uci.ics.texera.config.KubernetesConfig
+import edu.uci.ics.texera.dao.jooq.generated.enums.PrivilegeEnum
 import edu.uci.ics.texera.dao.jooq.generated.tables.pojos.User
 import org.apache.http.client.utils.URLEncodedUtils
 
@@ -29,6 +32,7 @@ import java.nio.charset.Charset
 import javax.websocket.HandshakeResponse
 import javax.websocket.server.{HandshakeRequest, ServerEndpointConfig}
 import scala.jdk.CollectionConverters.ListHasAsScala
+import scala.jdk.CollectionConverters._
 
 /**
   * This configurator extracts HTTPSession and associates it to ServerEndpointConfig,
@@ -46,30 +50,70 @@ class ServletAwareConfigurator extends ServerEndpointConfig.Configurator with La
       response: HandshakeResponse
   ): Unit = {
     try {
-      val params =
-        URLEncodedUtils.parse(new URI("?" + request.getQueryString), Charset.defaultCharset())
-      params.asScala
-        .map(pair => pair.getName -> pair.getValue)
-        .toMap
-        .get("access-token")
-        .map(token => {
-          val claims = jwtConsumer.process(token).getJwtClaims
-          config.getUserProperties.put(
-            classOf[User].getName,
-            new User(
-              claims.getClaimValue("userId").asInstanceOf[Long].toInt,
-              claims.getSubject,
-              String.valueOf(claims.getClaimValue("email").asInstanceOf[String]),
-              null,
-              null,
-              null,
-              null,
-              null,
-              null
-            )
-          )
-        })
+      val headers = request.getHeaders.asScala.view.mapValues(_.asScala.headOption).toMap
+      if (
+        headers.contains(HeaderField.UserComputingUnitAccess) &&
+        headers.contains(HeaderField.UserId) &&
+        headers.contains(HeaderField.UserName) &&
+        headers.contains(HeaderField.UserEmail)
+      ) {
+        // KUBERNETES MODE: Construct the User object from trusted headers
+        // coming from envoy and generated by access control service.
+
+        val userId = headers.get(HeaderField.UserId).flatten.map(_.toInt).get
+        val userName = headers.get(HeaderField.UserName).flatten.get
+        val userEmail = headers.get(HeaderField.UserEmail).flatten.get
+        val cuAccess = headers.get(HeaderField.UserComputingUnitAccess).flatten.getOrElse("")
+        config.getUserProperties.put(HeaderField.UserComputingUnitAccess, cuAccess)
+        logger.info(
+          s"User ID: $userId, User Name: $userName, User Email: $userEmail with CU Access: $cuAccess"
+        )
 
+        config.getUserProperties.put(
+          classOf[User].getName,
+          new User(
+            userId,
+            userName,
+            userEmail,
+            null,
+            null,
+            null,
+            null,
+            null,
+            null
+          )
+        )
+        logger.debug(s"User created from headers: ID=$userId, Name=$userName")
+      } else {
+        // SINGLE NODE MODE: Construct the User object from JWT in query parameters.
+        val params =
+          URLEncodedUtils.parse(new URI("?" + request.getQueryString), Charset.defaultCharset())
+        config.getUserProperties.put(
+          HeaderField.UserComputingUnitAccess,
+          PrivilegeEnum.WRITE.name()
+        )
+        params.asScala
+          .map(pair => pair.getName -> pair.getValue)
+          .toMap
+          .get("access-token")
+          .map(token => {
+            val claims = jwtConsumer.process(token).getJwtClaims
+            config.getUserProperties.put(
+              classOf[User].getName,
+              new User(
+                claims.getClaimValue("userId").asInstanceOf[Long].toInt,
+                claims.getSubject,
+                String.valueOf(claims.getClaimValue("email").asInstanceOf[String]),
+                null,
+                null,
+                null,
+                null,
+                null,
+                null
+              )
+            )
+          })
+      }
     } catch {
       case e: Exception =>
         logger.error("Failed to retrieve the User during websocket handshake", e)

core/amber/src/main/scala/edu/uci/ics/texera/web/SessionState.scala

@@ -20,6 +20,7 @@
 package edu.uci.ics.texera.web
 
 import edu.uci.ics.amber.util.JSONUtils.objectMapper
+import edu.uci.ics.texera.dao.jooq.generated.enums.PrivilegeEnum
 import edu.uci.ics.texera.web.model.websocket.event.TexeraWebSocketEvent
 import edu.uci.ics.texera.web.service.WorkflowService
 import io.reactivex.rxjava3.disposables.Disposable
@@ -53,6 +54,7 @@ class SessionState(session: Session) {
   private var currentWorkflowState: Option[WorkflowService] = None
   private var workflowSubscription = Disposable.empty()
   private var executionSubscription = Disposable.empty()
+  private var userComputingUnitAccess: PrivilegeEnum = PrivilegeEnum.NONE
 
   def send(msg: TexeraWebSocketEvent): Unit = {
     session.getAsyncRemote.sendText(objectMapper.writeValueAsString(msg))
@@ -80,4 +82,9 @@ class SessionState(session: Session) {
     )
 
   }
+
+  def setUserComputingUnitAccess(cuAccess: PrivilegeEnum): Unit = {
+    this.userComputingUnitAccess = cuAccess
+  }
+  def getUserComputingUnitAccess: PrivilegeEnum = this.userComputingUnitAccess
 }

core/amber/src/main/scala/edu/uci/ics/texera/web/resource/WorkflowWebsocketResource.scala

@@ -27,6 +27,8 @@ import edu.uci.ics.amber.error.ErrorUtils.getStackTraceWithAllCauses
 import edu.uci.ics.amber.core.virtualidentity.WorkflowIdentity
 import edu.uci.ics.amber.core.workflowruntimestate.FatalErrorType.COMPILATION_ERROR
 import edu.uci.ics.amber.core.workflowruntimestate.WorkflowFatalError
+import edu.uci.ics.texera.auth.util.HeaderField
+import edu.uci.ics.texera.dao.jooq.generated.enums.PrivilegeEnum
 import edu.uci.ics.texera.dao.jooq.generated.tables.pojos.User
 import edu.uci.ics.texera.web.model.websocket.event.{WorkflowErrorEvent, WorkflowStateEvent}
 import edu.uci.ics.texera.web.model.websocket.request._
@@ -51,13 +53,22 @@ class WorkflowWebsocketResource extends LazyLogging {
     SessionState.setState(session.getId, sessionState)
     val wid = session.getRequestParameterMap.get("wid").get(0).toLong
     val cuid = session.getRequestParameterMap.get("cuid").get(0).toInt
+    val cuAccessEnum: PrivilegeEnum = PrivilegeEnum.valueOf(
+      session.getUserProperties
+        .get(HeaderField.UserComputingUnitAccess)
+        .asInstanceOf[String]
+    )
+
+    sessionState.setUserComputingUnitAccess(cuAccessEnum)
+    logger.info(
+      s"Websocket connection opened for workflow $wid with computing unit $cuid and access $cuAccessEnum"
+    )
     // hack to refresh frontend run button state
     sessionState.send(WorkflowStateEvent("Uninitialized"))
     val workflowState =
       WorkflowService.getOrCreate(WorkflowIdentity(wid), cuid)
     sessionState.subscribe(workflowState)
     sessionState.send(ClusterStatusUpdateEvent(ClusterListener.numWorkerNodesInCluster))
-    logger.info("connection open")
   }
 
   @OnClose
@@ -94,6 +105,9 @@ class WorkflowWebsocketResource extends LazyLogging {
             sessionState.send(modifyLogicResponse)
           }
         case workflowExecuteRequest: WorkflowExecuteRequest =>
+          if (sessionState.getUserComputingUnitAccess != PrivilegeEnum.WRITE) {
+            throw new IllegalStateException("User does not have write access to the computing unit")
+          }
           workflowStateOpt match {
             case Some(workflow) =>
               sessionState.send(WorkflowStateEvent("Initializing"))

core/build.sbt

@@ -99,6 +99,7 @@ lazy val CoreProject = (project in file("."))
     DAO,
     Config,
     ConfigService,
+    AccessControlService,
     Auth,
     WorkflowCore,
     ComputingUnitManagingService,

core/gui/src/app/dashboard/component/user/share-access/share-access.component.html

@@ -115,9 +115,7 @@
     <br />
 
     <nz-card nzTitle="Share">
-      <div
-        style="height: 50px"
-        *ngIf="type !== 'computing-unit'">
+      <div style="height: 50px">
         Access Level:
         <select formControlName="accessLevel">
           <option value="READ">read</option>

core/gui/src/app/workspace/component/menu/menu.component.html

@@ -364,7 +364,7 @@
           [nzPopoverTrigger]="this.config.env.userSystemEnabled?'hover':null"
           [nzPopoverContent]="executionSettings"
           nzPopoverPlacement="bottom"
-          [disabled]="runDisable || (!workflowWebsocketService.isConnected && computingUnitStatus !== ComputingUnitState.NoComputingUnit) || displayParticularWorkflowVersion"
+          [disabled]="runDisable || (!workflowWebsocketService.isConnected && computingUnitStatus !== ComputingUnitState.NoComputingUnit) || displayParticularWorkflowVersion || selectedComputingUnit?.accessPrivilege !== Privilege.WRITE"
           id="run-button"
           nz-button
           nzType="primary">

core/gui/src/app/workspace/component/menu/menu.component.ts

@@ -54,6 +54,8 @@ import { ComputingUnitStatusService } from "../../service/computing-unit-status/
 import { ComputingUnitState } from "../../types/computing-unit-connection.interface";
 import { ComputingUnitSelectionComponent } from "../power-button/computing-unit-selection.component";
 import { GuiConfigService } from "../../../common/service/gui-config.service";
+import { DashboardWorkflowComputingUnit } from "../../types/workflow-computing-unit";
+import { Privilege } from "../../../dashboard/type/share-access.interface";
 
 /**
  * MenuComponent is the top level menu bar that shows
@@ -110,6 +112,7 @@ export class MenuComponent implements OnInit, OnDestroy {
 
   // Computing unit status variables
   private computingUnitStatusSubscription: Subscription = new Subscription();
+  public selectedComputingUnit: DashboardWorkflowComputingUnit | null = null;
   public computingUnitStatus: ComputingUnitState = ComputingUnitState.NoComputingUnit;
 
   @ViewChild(ComputingUnitSelectionComponent) computingUnitSelectionComponent!: ComputingUnitSelectionComponent;
@@ -162,7 +165,8 @@ export class MenuComponent implements OnInit, OnDestroy {
     this.registerWorkflowModifiableChangedHandler();
     this.registerWorkflowIdUpdateHandler();
 
-    // Subscribe to computing unit status changes
+    // Subscribe to computing unit
+    this.subscribeToComputingUnitSelection();
     this.subscribeToComputingUnitStatus();
   }
 
@@ -202,6 +206,15 @@ export class MenuComponent implements OnInit, OnDestroy {
     this.computingUnitStatusSubscription.unsubscribe();
   }
 
+  private subscribeToComputingUnitSelection(): void {
+    this.computingUnitStatusService
+      .getSelectedComputingUnit()
+      .pipe(untilDestroyed(this))
+      .subscribe(unit => {
+        this.selectedComputingUnit = unit;
+      });
+  }
+
   /**
    * Subscribe to computing unit status changes from the ComputingUnitStatusService
    */
@@ -720,4 +733,6 @@ export class MenuComponent implements OnInit, OnDestroy {
       this.config.env.workflowEmailNotificationEnabled && this.config.env.userSystemEnabled
     );
   }
+
+  protected readonly Privilege = Privilege;
 }

deployment/access-control-service.dockerfile

@@ -0,0 +1,54 @@
+# 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.
+
+FROM sbtscala/scala-sbt:eclipse-temurin-jammy-11.0.17_8_1.9.3_2.13.11 AS build
+
+# Set working directory
+WORKDIR /core
+
+# Copy all projects under core to /core
+COPY core/ .
+
+# Update system and install dependencies
+RUN apt-get update && apt-get install -y \
+    netcat \
+    unzip \
+    libpq-dev \
+    && apt-get clean
+
+WORKDIR /core
+# Add .git for runtime calls to jgit from OPversion
+COPY .git ../.git
+
+RUN sbt clean AccessControlService/dist
+
+# Unzip the texera binary
+RUN unzip  access-control-service/target/universal/access-control-service-*.zip -d target/
+
+FROM eclipse-temurin:11-jre-jammy AS runtime
+
+WORKDIR /core
+
+COPY --from=build /.git /.git
+# Copy the built texera binary from the build phase
+COPY --from=build /core/target/access-control-service* /core/
+# Copy resources directories under /core from build phase
+COPY --from=build /core/access-control-service/src/main/resources /core/access-control-service/src/main/resources
+
+CMD ["bin/access-control-service"]
+
+EXPOSE 9096

deployment/k8s/texera-helmchart/templates/access-control-service-deployment.yaml

@@ -0,0 +1,60 @@
+# 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.
+
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name:  {{.Release.Name}}-{{ .Values.accessControlService.name }}
+  namespace: {{ .Release.Namespace }}
+  labels:
+    app: {{ .Release.Name }}-{{ .Values.accessControlService.name }}
+spec:
+  replicas: {{ .Values.accessControlService.numOfPods | default 1 }}
+  selector:
+    matchLabels:
+      app: {{ .Release.Name }}-{{ .Values.accessControlService.name }}
+  template:
+    metadata:
+      labels:
+        app: {{ .Release.Name }}-{{ .Values.accessControlService.name }}
+    spec:
+      containers:
+        - name: {{ .Values.accessControlService.name }}
+          image: {{ .Values.accessControlService.imageName }}
+          imagePullPolicy: {{ .Values.texeraImages.pullPolicy }}
+          ports:
+            - containerPort: {{ .Values.accessControlService.service.port }}
+          env:
+            - name: STORAGE_JDBC_URL
+              value: jdbc:postgresql://{{ .Release.Name }}-postgresql:5432/texera_db?currentSchema=texera_db,public
+            - name: STORAGE_JDBC_PASSWORD
+              valueFrom:
+                secretKeyRef:
+                  name: {{ .Release.Name }}-postgresql
+                  key: postgres-password
+          livenessProbe:
+            httpGet:
+              path: /api/healthcheck
+              port: {{ .Values.accessControlService.service.port }}
+            initialDelaySeconds: 30
+            periodSeconds: 10
+          readinessProbe:
+            httpGet:
+              path: /api/healthcheck
+              port: {{ .Values.accessControlService.service.port }}
+            initialDelaySeconds: 5
+            periodSeconds: 5