Improve session REST security

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

@@ -3336,6 +3336,21 @@ object KyuubiConf {
       .regexConf
       .createOptional
 
+  val SESSION_CONF_DISPLAY_MODE: ConfigEntry[String] =
+    buildConf("kyuubi.session.conf.display.mode")
+      .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 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,28 @@ 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_SECRET_REDACTION_PATTERN, SESSION_CONF_DISPLAY_MODE}
 import org.apache.kyuubi.ha.client.ServiceNodeInfo
 import org.apache.kyuubi.operation.KyuubiOperation
 import org.apache.kyuubi.session.KyuubiSession
 
 object ApiUtils extends Logging {
+
+  private def buildConf(
+      rawConf: Map[String, String],
+      session: KyuubiSession): java.util.Map[String, String] = {
+    session.sessionManager.getConf.get(SESSION_CONF_DISPLAY_MODE) match {
+      case "NONE" => Map.empty[String, String].asJava
+      case "ORIGINAL" => rawConf.asJava
+      case _ =>
+        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 +52,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 +63,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

@@ -52,11 +52,15 @@ 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 userName = fe.getSessionUser(Map.empty[String, String])
+    sessionManager
+      .allSessions()
+      .filter(session => session.user == userName)
+      .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,73 @@ 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.SESSION_CONF_DISPLAY_MODE, mode)
+    try f finally conf.set(KyuubiConf.SESSION_CONF_DISPLAY_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)")
   }
 
 }