Address comments

Single auth checkpoint for every current and future RPC. Dropped the redundant calls in Controller.

Add comment for remote storages.

Harden validateAppId

Author: afterincomparableyum

common/src/main/scala/org/apache/celeborn/common/rpc/RpcEndpoint.scala

@@ -20,6 +20,7 @@ package org.apache.celeborn.common.rpc
 import org.apache.celeborn.common.exception.CelebornException
 import org.apache.celeborn.common.network.client.TransportClient
 import org.apache.celeborn.common.rpc.netty.RemoteNettyRpcCallContext
+import org.apache.celeborn.common.util.Utils
 
 /**
  * A factory class to create the [[RpcEnv]]. It must have an empty constructor so that it can be
@@ -138,6 +139,11 @@ trait RpcEndpoint {
   }
 
   def checkAuth(context: RpcCallContext, appId: String): Unit = {
+    // Validate the application id at the single auth chokepoint so every current
+    // and future RPC handler that calls checkAuth is covered, and so it runs even
+    // when auth is disabled (clientId == null). This guards the worker against
+    // path traversal via appId (e.g. "../foo") before any filesystem path is built.
+    Utils.validateAppId(appId)
     context match {
       case remoteContext: RemoteNettyRpcCallContext =>
         checkAuth(remoteContext.transportClient, appId)

common/src/main/scala/org/apache/celeborn/common/util/Utils.scala

@@ -700,11 +700,12 @@ object Utils extends Logging {
     (appId, shuffleId)
   }
 
-  private val appIdPattern = "^[A-Za-z0-9_-]+$".r
+  private val appIdPattern = "[A-Za-z0-9_-]+".r.pattern
 
   def validateAppId(applicationId: String): Unit = {
-    if (applicationId == null || applicationId.isEmpty ||
-      appIdPattern.findFirstIn(applicationId).isEmpty) {
+    // matches() anchors the whole input, so a trailing newline (which `$` would
+    // otherwise tolerate) is rejected along with any other traversal character.
+    if (applicationId == null || !appIdPattern.matcher(applicationId).matches()) {
       throw new IllegalArgumentException(
         s"Invalid application id: '$applicationId'. " +
           "Application id must be non-empty and match [A-Za-z0-9_-]+.")

common/src/test/scala/org/apache/celeborn/common/util/UtilsSuite.scala

@@ -306,6 +306,8 @@ class UtilsSuite extends CelebornFunSuite {
       "app/id",
       "app\\id",
       "app id",
+      "app\n",
+      "valid_app\n",
       "",
       null).foreach { id =>
       intercept[IllegalArgumentException] {

worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Controller.scala

@@ -115,7 +115,6 @@ private[deploy] class Controller(
           pushDataTimeout,
           partitionSplitEnabled,
           isSegmentGranularityVisible) =>
-      Utils.validateAppId(applicationId)
       checkAuth(context, applicationId)
       val shuffleKey = Utils.makeShuffleKey(applicationId, shuffleId)
       workerSource.sample(WorkerSource.RESERVE_SLOTS_TIME, shuffleKey) {
@@ -147,7 +146,6 @@ private[deploy] class Controller(
           mapAttempts,
           epoch,
           mockFailure) =>
-      Utils.validateAppId(applicationId)
       checkAuth(context, applicationId)
       val shuffleKey = Utils.makeShuffleKey(applicationId, shuffleId)
       logDebug(s"Received CommitFiles request, $shuffleKey, primary files" +
@@ -166,9 +164,7 @@ private[deploy] class Controller(
         s"$commitFilesTimeMs ms.")
 
     case DestroyWorkerSlots(shuffleKey, primaryLocations, replicaLocations, mockFailure) =>
-      val applicationId = Utils.splitShuffleKey(shuffleKey)._1
-      Utils.validateAppId(applicationId)
-      checkAuth(context, applicationId)
+      checkAuth(context, Utils.splitShuffleKey(shuffleKey)._1)
       handleDestroy(context, shuffleKey, primaryLocations, replicaLocations, mockFailure)
   }
 

worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/StorageManager.scala

@@ -1161,6 +1161,9 @@ final private[worker] class StorageManager(conf: CelebornConf, workerSource: Abs
         throw new IOException(s"No available disks! suggested mountPoint $suggestedMountPoint")
       }
 
+      // NOTE: the DFS branches below (HDFS/S3/OSS) also build "$appId/$shuffleId"
+      // paths but rely solely on the upstream Utils.validateAppId guard at the RPC
+      // entry points
       if (storageType == Type.HDFS && location.getStorageInfo.HDFSAvailable()) {
         val shuffleDir =
           new Path(new Path(hdfsDir, conf.workerWorkingDir), s"$appId/$shuffleId")