Improve session REST security

docs/deployment/migration-guide.md

@@ -21,6 +21,8 @@
 
 * Since Kyuubi 1.12, the support of variable `<KYUUBI_HOME>` substitution in config `kyuubi.metadata.store.jdbc.url` is deprecated, use `{{KYUUBI_HOME}}` instead.
 * Since Kyuubi 1.12, default value of `kyuubi.metrics.json.location` is changed to `{{KYUUBI_HOME}}/metrics`, to restore previous behavior, change it to `{{KYUUBI_WORK_DIR_ROOT}}/metrics`.
+* Since Kyuubi 1.12, session configurations in REST API responses are redacted by default using `kyuubi.server.redaction.regex`. Use `kyuubi.server.conf.retrieveMode` to control this behavior: `REDACTED` (default), `ORIGINAL` (no redaction), or `NONE` (omit configs entirely).
+* Since Kyuubi 1.12, `GET /api/v1/sessions` returns only sessions owned by the authenticated user instead of all sessions on the server. To restore the previous behavior, set `kyuubi.frontend.rest.legacy.v1.sessionsReturnAllUsers=true`.
 * Since Kyuubi 1.12, the configuration `spark.sql.kyuubi.hive.connector.dropTableAsPurgeTable` is introduced by Kyuubi Spark Hive connector(KSHC) to control whether DROP TABLE command completely remove its data by skipping HDFS trash. The default value is false. To restore the legacy behavior, set it to true.
 
 ## Upgrading from Kyuubi 1.10 to 1.11

kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala

@@ -3336,6 +3336,32 @@ object KyuubiConf {
       .regexConf
       .createOptional
 
+  val SERVER_CONF_RETRIEVE_MODE: ConfigEntry[String] =
+    buildConf("kyuubi.server.conf.retrieveMode")
+      .serverOnly
+      .doc("Controls how session configurations are returned in REST API responses. " +
+        "Supported values: " +
+        "<ul>" +
+        "<li>REDACTED: Mask values that match kyuubi.server.redaction.regex (default).</li>" +
+        "<li>ORIGINAL: Return the raw config values as-is.</li>" +
+        "<li>NONE: Omit the conf map from responses entirely.</li>" +
+        "</ul>")
+      .version("1.12.0")
+      .stringConf
+      .checkValues(Set("REDACTED", "ORIGINAL", "NONE"))
+      .createWithDefault("REDACTED")
+
+  val FRONTEND_REST_SESSION_LIST_LEGACY_MODE: ConfigEntry[Boolean] =
+    buildConf("kyuubi.frontend.rest.legacy.v1.sessionsReturnAllUsers")
+      .serverOnly
+      .doc("When true, GET /api/v1/sessions returns all sessions on the server regardless " +
+        "of the calling user (legacy behavior). When false (default), only sessions owned " +
+        "by the authenticated user are returned. " +
+        "This flag is provided for backward compatibility and will be removed in a future release.")
+      .version("1.12.0")
+      .booleanConf
+      .createWithDefault(false)
+
   val SERVER_PERIODIC_GC_INTERVAL: ConfigEntry[Long] =
     buildConf("kyuubi.server.periodicGC.interval")
       .doc("How often to trigger the periodic garbage collection. 0 will disable it.")

kyuubi-server/src/main/scala/org/apache/kyuubi/server/api/ApiUtils.scala

@@ -22,13 +22,33 @@ import scala.collection.JavaConverters._
 import org.apache.kyuubi.{Logging, Utils}
 import org.apache.kyuubi.client.api.v1.dto
 import org.apache.kyuubi.client.api.v1.dto.{OperationData, OperationProgress, ServerData, SessionData}
+import org.apache.kyuubi.config.KyuubiConf.{SERVER_CONF_RETRIEVE_MODE, SERVER_SECRET_REDACTION_PATTERN}
 import org.apache.kyuubi.ha.client.ServiceNodeInfo
 import org.apache.kyuubi.operation.KyuubiOperation
 import org.apache.kyuubi.session.KyuubiSession
 
+object ConfRetrieveMode extends Enumeration {
+  val REDACTED, ORIGINAL, NONE = Value
+}
+
 object ApiUtils extends Logging {
+
+  private def buildConf(
+      rawConf: Map[String, String],
+      session: KyuubiSession): java.util.Map[String, String] = {
+    ConfRetrieveMode.withName(
+      session.sessionManager.getConf.get(SERVER_CONF_RETRIEVE_MODE)) match {
+      case ConfRetrieveMode.NONE => Map.empty[String, String].asJava
+      case ConfRetrieveMode.ORIGINAL => rawConf.asJava
+      case ConfRetrieveMode.REDACTED =>
+        val pattern = session.sessionManager.getConf.get(SERVER_SECRET_REDACTION_PATTERN)
+        Utils.redact(pattern, rawConf.toSeq).toMap.asJava
+    }
+  }
+
   def sessionEvent(session: KyuubiSession): dto.KyuubiSessionEvent = {
-    session.getSessionEvent.map(event =>
+    session.getSessionEvent.map { event =>
+      val conf = buildConf(event.conf, session)
       dto.KyuubiSessionEvent.builder()
         .sessionId(event.sessionId)
         .clientVersion(event.clientVersion)
@@ -37,7 +57,7 @@ object ApiUtils extends Logging {
         .user(event.user)
         .clientIp(event.clientIP)
         .serverIp(event.serverIP)
-        .conf(event.conf.asJava)
+        .conf(conf)
         .remoteSessionId(event.remoteSessionId)
         .engineId(event.engineId)
         .engineName(event.engineName)
@@ -48,17 +68,19 @@ object ApiUtils extends Logging {
         .endTime(event.endTime)
         .totalOperations(event.totalOperations)
         .exception(event.exception.orNull)
-        .build()).orNull
+        .build()
+    }.orNull
   }
 
   def sessionData(session: KyuubiSession): SessionData = {
     val sessionEvent = session.getSessionEvent
+    val conf = buildConf(session.conf, session)
     new SessionData(
       session.handle.identifier.toString,
       sessionEvent.map(_.remoteSessionId).getOrElse(""),
       session.user,
       session.ipAddress,
-      session.conf.asJava,
+      conf,
       session.createTime,
       session.lastAccessTime - session.createTime,
       session.getNoOperationTime,

kyuubi-server/src/main/scala/org/apache/kyuubi/server/api/v1/SessionsResource.scala

@@ -32,6 +32,7 @@ import org.apache.commons.lang3.StringUtils
 import org.apache.kyuubi.Logging
 import org.apache.kyuubi.client.api.v1.dto
 import org.apache.kyuubi.client.api.v1.dto._
+import org.apache.kyuubi.config.KyuubiConf.FRONTEND_REST_SESSION_LIST_LEGACY_MODE
 import org.apache.kyuubi.config.KyuubiReservedKeys._
 import org.apache.kyuubi.operation.{KyuubiOperation, OperationHandle}
 import org.apache.kyuubi.server.api.{ApiRequestContext, ApiUtils}
@@ -52,11 +53,18 @@ private[v1] class SessionsResource extends ApiRequestContext with Logging {
     content = Array(new Content(
       mediaType = MediaType.APPLICATION_JSON,
       array = new ArraySchema(schema = new Schema(implementation = classOf[SessionData])))),
-    description = "get the list of all live sessions")
+    description = "get the list of live sessions for the current user")
   @GET
   def sessions(): Seq[SessionData] = {
-    sessionManager.allSessions()
-      .map(session => ApiUtils.sessionData(session.asInstanceOf[KyuubiSession])).toSeq
+    val legacyMode = sessionManager.getConf.get(FRONTEND_REST_SESSION_LIST_LEGACY_MODE)
+    val allSessions = sessionManager.allSessions()
+    val filtered =
+      if (legacyMode) allSessions
+      else {
+        val userName = fe.getSessionUser(Map.empty[String, String])
+        allSessions.filter(session => session.user == userName)
+      }
+    filtered.map(session => ApiUtils.sessionData(session.asInstanceOf[KyuubiSession])).toSeq
   }
 
   @ApiResponse(

kyuubi-server/src/test/scala/org/apache/kyuubi/server/api/v1/SessionsResourceSuite.scala

@@ -40,6 +40,12 @@ import org.apache.kyuubi.session.SessionType
 
 class SessionsResourceSuite extends KyuubiFunSuite with RestFrontendTestHelper {
 
+  override protected lazy val conf: KyuubiConf = {
+    val c = KyuubiConf()
+    c.set(KyuubiConf.SERVER_SECRET_REDACTION_PATTERN, "(?i)password".r)
+    c
+  }
+
   override protected def beforeEach(): Unit = {
     super.beforeEach()
     eventually(timeout(10.seconds), interval(200.milliseconds)) {
@@ -389,4 +395,74 @@ class SessionsResourceSuite extends KyuubiFunSuite with RestFrontendTestHelper {
     assert(operations.size == 1)
     assert(sessionHandle.toString.equals(operations.head.getSessionId))
   }
+
+  test("get /sessions returns redacted spark confs when mode is REDACTED") {
+    val sensitiveKey = "spark.password"
+    val sensitiveValue = "superSecret123"
+    val requestObj = new SessionOpenRequest(Map(sensitiveKey -> sensitiveValue).asJava)
+
+    val response = webTarget.path("api/v1/sessions")
+      .request(MediaType.APPLICATION_JSON_TYPE)
+      .post(Entity.entity(requestObj, MediaType.APPLICATION_JSON_TYPE))
+    assert(200 == response.getStatus)
+    val sessionHandle = response.readEntity(classOf[SessionHandle]).getIdentifier
+
+    val response2 = webTarget.path("api/v1/sessions").request().get()
+    assert(200 == response2.getStatus)
+    val sessions = response2.readEntity(new GenericType[Seq[SessionData]]() {})
+    val sessionConf = sessions.find(_.getIdentifier == sessionHandle.toString).get.getConf
+
+    assert(sessionConf.get(sensitiveKey) != sensitiveValue)
+    assert(sessionConf.get(sensitiveKey) == "*********(redacted)")
+
+    val delResp = webTarget.path(s"api/v1/sessions/$sessionHandle").request().delete()
+    assert(200 == delResp.getStatus)
+  }
+
+  test("get /sessions returns empty conf when mode is NONE") {
+    withSessionConfDisplayMode("NONE") {
+      val requestObj =
+        new SessionOpenRequest(Map("spark.password" -> "secret", "key" -> "val").asJava)
+      val r = webTarget.path("api/v1/sessions")
+        .request(MediaType.APPLICATION_JSON_TYPE)
+        .post(Entity.entity(requestObj, MediaType.APPLICATION_JSON_TYPE))
+      assert(200 == r.getStatus)
+      val sessionHandle = r.readEntity(classOf[SessionHandle]).getIdentifier
+
+      val r2 = webTarget.path("api/v1/sessions").request().get()
+      assert(200 == r2.getStatus)
+      val sessions = r2.readEntity(new GenericType[Seq[SessionData]]() {})
+      val sessionConf = sessions.find(_.getIdentifier == sessionHandle.toString).get.getConf
+      assert(sessionConf.isEmpty)
+
+      webTarget.path(s"api/v1/sessions/$sessionHandle").request().delete()
+    }
+  }
+
+  test("get /sessions returns raw conf when mode is ORIGINAL") {
+    withSessionConfDisplayMode("ORIGINAL") {
+      val sensitiveKey = "spark.password"
+      val sensitiveValue = "plainVisible"
+      val requestObj = new SessionOpenRequest(Map(sensitiveKey -> sensitiveValue).asJava)
+      val r = webTarget.path("api/v1/sessions")
+        .request(MediaType.APPLICATION_JSON_TYPE)
+        .post(Entity.entity(requestObj, MediaType.APPLICATION_JSON_TYPE))
+      assert(200 == r.getStatus)
+      val sessionHandle = r.readEntity(classOf[SessionHandle]).getIdentifier
+
+      val r2 = webTarget.path("api/v1/sessions").request().get()
+      assert(200 == r2.getStatus)
+      val sessions = r2.readEntity(new GenericType[Seq[SessionData]]() {})
+      val sessionConf = sessions.find(_.getIdentifier == sessionHandle.toString).get.getConf
+      assert(sessionConf.get(sensitiveKey) == sensitiveValue)
+
+      webTarget.path(s"api/v1/sessions/$sessionHandle").request().delete()
+    }
+  }
+
+  private def withSessionConfDisplayMode(mode: String)(f: => Unit): Unit = {
+    conf.set(KyuubiConf.SERVER_CONF_RETRIEVE_MODE, mode)
+    try f
+    finally conf.set(KyuubiConf.SERVER_CONF_RETRIEVE_MODE, "REDACTED")
+  }
 }

kyuubi-server/src/test/scala/org/apache/kyuubi/server/rest/client/SessionCtlSuite.scala

@@ -38,13 +38,13 @@ class SessionCtlSuite extends RestClientTestHelper with TestPrematureExit {
   test("list sessions") {
     fe.be.sessionManager.openSession(
       TProtocolVersion.findByValue(1),
-      "admin",
+      clientPrincipalUser,
       "123456",
       "localhost",
       Map("testConfig" -> "testValue"))
 
     val args = Array("list", "session", "--authSchema", "spnego")
-    testPrematureExitForControlCli(args, "Session List (total 1)")
+    testPrematureExitForControlCli(args, "Live Session List (total 1)")
   }
 
 }