Improve access path validation

kyuubi-server/src/main/scala/org/apache/kyuubi/engine/KyuubiApplicationManager.scala

@@ -176,11 +176,15 @@ object KyuubiApplicationManager {
       appConf: Map[String, String],
       kyuubiConf: KyuubiConf): Unit = {
     if (kyuubiConf.get(KyuubiConf.SESSION_LOCAL_DIR_ALLOW_LIST).nonEmpty) {
-      (SparkProcessBuilder.PATH_CONFIGS.toSet ++
-        kyuubiConf.get(KyuubiConf.SERVER_SPARK_FILE_CONFIG_LIST)).flatMap { key =>
-        appConf.get(key).map(_.split(",")).getOrElse(Array.empty)
-      }.filter(_.nonEmpty).foreach { path =>
-        checkApplicationAccessPath(path, kyuubiConf)
+      val policedKeys = SparkProcessBuilder.PATH_CONFIGS.toSet ++
+        kyuubiConf.get(KyuubiConf.SERVER_SPARK_FILE_CONFIG_LIST)
+          .map(SparkProcessBuilder.convertConfigKey)
+      appConf.foreach { case (key, value) =>
+        if (policedKeys.contains(SparkProcessBuilder.convertConfigKey(key))) {
+          value.split(",").filter(_.nonEmpty).foreach { path =>
+            checkApplicationAccessPath(path, kyuubiConf)
+          }
+        }
       }
     }
   }

kyuubi-server/src/main/scala/org/apache/kyuubi/engine/spark/SparkProcessBuilder.scala

@@ -101,13 +101,7 @@ class SparkProcessBuilder(
    * - Otherwise, the key will be added a `spark.` prefix
    */
   protected def convertConfigKey(key: String): String = {
-    if (key.startsWith("spark.")) {
-      key
-    } else if (key.startsWith("hadoop.")) {
-      "spark.hadoop." + key
-    } else {
-      "spark." + key
-    }
+    SparkProcessBuilder.convertConfigKey(key)
   }
 
   private[kyuubi] def extractSparkCoreScalaVersion(fileNames: Iterable[String]): String = {
@@ -408,6 +402,16 @@ class SparkProcessBuilder(
 }
 
 object SparkProcessBuilder {
+  private[kyuubi] def convertConfigKey(key: String): String = {
+    if (key.startsWith("spark.")) {
+      key
+    } else if (key.startsWith("hadoop.")) {
+      "spark.hadoop." + key
+    } else {
+      "spark." + key
+    }
+  }
+
   final val APP_KEY = "spark.app.name"
   final val TAG_KEY = "spark.yarn.tags"
   final val MASTER_KEY = "spark.master"

kyuubi-server/src/test/scala/org/apache/kyuubi/engine/KyuubiApplicationManagerSuite.scala

@@ -20,6 +20,7 @@ package org.apache.kyuubi.engine
 import org.apache.kyuubi.{KyuubiException, KyuubiFunSuite}
 import org.apache.kyuubi.config.KyuubiConf
 import org.apache.kyuubi.engine.KubernetesApplicationOperation.LABEL_KYUUBI_UNIQUE_KEY
+import org.apache.kyuubi.engine.spark.SparkProcessBuilder
 
 class KyuubiApplicationManagerSuite extends KyuubiFunSuite {
   test("application access path") {
@@ -115,6 +116,37 @@ class KyuubiApplicationManagerSuite extends KyuubiFunSuite {
     localDirLimitConf.unset(KyuubiConf.SERVER_SPARK_FILE_CONFIG_LIST)
   }
 
+  test("application access path checks unprefixed spark file conf aliases") {
+    val localDirLimitConf = KyuubiConf()
+      .set(KyuubiConf.SESSION_LOCAL_DIR_ALLOW_LIST, Set("/apache/kyuubi"))
+
+    SparkProcessBuilder.PATH_CONFIGS.foreach { canonicalKey =>
+      Seq(canonicalKey, canonicalKey.stripPrefix("spark.")).foreach { key =>
+        val e = intercept[KyuubiException] {
+          KyuubiApplicationManager.checkApplicationAccessPaths(
+            "SPARK",
+            Map(key -> "/etc/passwd"),
+            localDirLimitConf)
+        }
+        assert(e.getMessage.contains("is not in the local dir allow list"), s"key=$key")
+      }
+    }
+
+    Seq("spark.files", "files").foreach { key =>
+      KyuubiApplicationManager.checkApplicationAccessPaths(
+        "SPARK",
+        Map(key -> "/apache/kyuubi/a.jar"),
+        localDirLimitConf)
+    }
+
+    intercept[KyuubiException] {
+      KyuubiApplicationManager.checkApplicationAccessPaths(
+        "SPARK",
+        Map("spark.files" -> "/apache/kyuubi/a.jar", "files" -> "/etc/passwd"),
+        localDirLimitConf)
+    }
+  }
+
   test("Test kyuubi application Manager tag spark on kubernetes application") {
     val conf: KyuubiConf = KyuubiConf()
     val tag = "kyuubi-test-tag"