[SPARK-57480][SQL] Stash allowSettingSystemProperties on a static field to prevent set:hiveconf bypass

sql/hive-thriftserver/src/main/java/org/apache/hive/service/cli/session/HiveSessionImpl.java

@@ -234,6 +234,19 @@ private void configureSession(Map<String, String> sessionConfMap) throws HiveSQL
     }
   }
 
+  // Resolved once at server init by SparkSQLSessionManager. See setAllowSettingSystemProperties.
+  private static volatile boolean allowSettingSystemProperties = false;
+
+  /**
+   * Configure whether `set:system:*` is permitted. Intended to be called once at server init
+   * from SparkSQLSessionManager. Reading and writing this value is intentionally not routed
+   * through any per-session HiveConf, since that would let a low-privilege client mutate it
+   * via `set:hiveconf:` from the same JDBC overlay this gate is meant to guard.
+   */
+  public static void setAllowSettingSystemProperties(boolean allow) {
+    allowSettingSystemProperties = allow;
+  }
+
   // Copy from org.apache.hadoop.hive.ql.processors.SetProcessor, only changes:
   // 1. setConf(varname, propName, varvalue, true) when varname.startsWith(HIVECONF_PREFIX)
   // 2. system:* variables can not be set unless the legacy configuration
@@ -251,8 +264,9 @@ public static int setVariable(String varname, String varvalue) throws Exception
     } else if (varname.startsWith(SYSTEM_PREFIX)){
       // Setting JVM system properties is allowed only when the legacy configuration
       // spark.sql.legacy.hive.thriftServer.allowSettingSystemProperties is enabled.
-      if (ss.getConf().getBoolean(
-          "spark.sql.legacy.hive.thriftServer.allowSettingSystemProperties", false)) {
+      // The toggle is read from a static field set once at server init time, NOT from the
+      // per-session HiveConf, to prevent bypass via `set:hiveconf:`.
+      if (allowSettingSystemProperties) {
         String propName = varname.substring(SYSTEM_PREFIX.length());
         System.getProperties().setProperty(propName, substitution.substitute(ss.getConf(),varvalue));
       } else {

sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkSQLSessionManager.scala

@@ -21,7 +21,7 @@ import scala.util.control.NonFatal
 
 import org.apache.hadoop.hive.conf.HiveConf
 import org.apache.hive.service.cli.SessionHandle
-import org.apache.hive.service.cli.session.SessionManager
+import org.apache.hive.service.cli.session.{HiveSessionImpl, SessionManager}
 import org.apache.hive.service.rpc.thrift.TProtocolVersion
 import org.apache.hive.service.server.HiveServer2
 
@@ -40,10 +40,10 @@ private[hive] class SparkSQLSessionManager(hiveServer: HiveServer2, sparkSession
 
   override def init(hiveConf: HiveConf): Unit = {
     setSuperField(this, "operationManager", sparkSqlOperationManager)
-    // Propagate the legacy toggle into HiveConf so that HiveSessionImpl.setVariable,
-    // which runs during session open before the SparkSession is attached, can read it.
-    hiveConf.setBoolean(
-      StaticSQLConf.LEGACY_HIVE_THRIFT_SERVER_ALLOW_SETTING_SYSTEM_PROPERTIES.key,
+    // Stash the toggle on a static field instead of writing it into HiveConf. Storing it on
+    // per-session HiveConf would let a JDBC client flip it from the same `set:` overlay the
+    // system: gate is meant to guard (via `set:hiveconf:<key>=true`).
+    HiveSessionImpl.setAllowSettingSystemProperties(
       sparkSession.sessionState.conf.getConf(
         StaticSQLConf.LEGACY_HIVE_THRIFT_SERVER_ALLOW_SETTING_SYSTEM_PROPERTIES))
     super.init(hiveConf)

sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/HiveSessionImplSuite.scala

@@ -66,18 +66,20 @@ class HiveSessionImplSuite extends SparkFunSuite {
   }
 
   private def withSystemPropSession[T](allowSettingSystemProperties: Boolean)(f: => T): T = {
+    HiveSessionImpl.setAllowSettingSystemProperties(allowSettingSystemProperties)
     val conf = new HiveConf()
-    conf.setBoolean(
-      StaticSQLConf.LEGACY_HIVE_THRIFT_SERVER_ALLOW_SETTING_SYSTEM_PROPERTIES.key,
-      allowSettingSystemProperties)
     val sessionImpl = new HiveSessionImpl(
       TProtocolVersion.HIVE_CLI_SERVICE_PROTOCOL_V1, "", "", conf, "")
     sessionImpl.setSessionManager(new SessionManager(null))
     sessionImpl.setOperationManager(new OperationManagerMock())
     sessionImpl.open(Map.empty[String, String].asJava)
     // open() does not call SessionState.start(), so err is not initialized in this unit test.
     SessionState.get().err = new PrintStream(new ByteArrayOutputStream())
-    try f finally sessionImpl.close()
+    try f finally {
+      sessionImpl.close()
+      // Restore default-deny so other tests don't inherit the relaxed state.
+      HiveSessionImpl.setAllowSettingSystemProperties(false)
+    }
   }
 
   test("SPARK-57441: system:* variables can not be set by default") {
@@ -105,6 +107,27 @@ class HiveSessionImplSuite extends SparkFunSuite {
       System.clearProperty(key)
     }
   }
+
+  test("SPARK-57480: set:hiveconf cannot flip the legacy flag mid-session to bypass " +
+      "the system:* gate") {
+    val key = "spark.test.HiveSessionImplSuite.bypassRegression"
+    val flagKey = StaticSQLConf.LEGACY_HIVE_THRIFT_SERVER_ALLOW_SETTING_SYSTEM_PROPERTIES.key
+    try {
+      withSystemPropSession(allowSettingSystemProperties = false) {
+        // The system: gate must not read its toggle from the per-session HiveConf, since
+        // `set:hiveconf:` directives write to that same HiveConf. Verify a client cannot
+        // flip the flag mid-session via `set:hiveconf:<flag-key>=true` and then bypass
+        // the gate. The hiveconf write itself is permitted (no-op for the gate).
+        HiveSessionImpl.setVariable("hiveconf:" + flagKey, "true")
+        val rcSystem = HiveSessionImpl.setVariable("system:" + key, "value")
+        assert(rcSystem == 1, s"system:* should still be rejected, got rc=$rcSystem")
+        assert(System.getProperty(key) == null,
+          s"system:* should not have been set, got ${System.getProperty(key)}")
+      }
+    } finally {
+      System.clearProperty(key)
+    }
+  }
 }
 
 class OperationManagerMock extends OperationManager {