[KYUUBI #7270]Enable BasicAuth for Prometheus metrics endpoint

kyuubi-metrics/src/main/scala/org/apache/kyuubi/metrics/BasicAuthFilter.scala

@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kyuubi.metrics
+
+import java.nio.charset.StandardCharsets
+import java.security.MessageDigest
+import java.util.Base64
+import javax.servlet._
+import javax.servlet.http.{HttpServletRequest, HttpServletResponse}
+
+import org.apache.kyuubi.Logging
+
+/**
+ * A servlet filter that implements HTTP Basic Authentication
+ */
+class BasicAuthFilter(username: String, password: String)
+  extends Filter with Logging {
+
+  private val credentials = s"$username:$password"
+  private val encodedCredentials =
+    Base64.getEncoder.encodeToString(credentials.getBytes(StandardCharsets.UTF_8))
+
+  override def init(filterConfig: FilterConfig): Unit = {
+    info("BasicAuthFilter initialized for Prometheus metrics endpoint")
+  }
+
+  override def doFilter(
+      request: ServletRequest,
+      response: ServletResponse,
+      chain: FilterChain): Unit = {
+
+    val httpRequest = request.asInstanceOf[HttpServletRequest]
+    val httpResponse = response.asInstanceOf[HttpServletResponse]
+
+    val authHeader = httpRequest.getHeader("Authorization")
+
+    if (authHeader != null && authHeader.startsWith("Basic ")) {
+      val providedCredentials = authHeader.substring(6) // Remove "Basic " prefix
+
+      if (isValidCredentials(providedCredentials)) {
+        // Authentication successful, continue the chain
+        chain.doFilter(request, response)
+      } else {
+        // Invalid credentials
+        warn(s"Invalid credentials provided from ${httpRequest.getRemoteAddr}")
+        sendUnauthorizedResponse(httpResponse)
+      }
+    } else {
+      // No credentials provided
+      debug(s"No credentials provided from ${httpRequest.getRemoteAddr}")
+      sendUnauthorizedResponse(httpResponse)
+    }
+  }
+
+  override def destroy(): Unit = {
+    info("BasicAuthFilter destroyed")
+  }
+
+  private def isValidCredentials(providedCredentials: String): Boolean = {
+    try {
+      // Use Java's constant-time comparison to prevent timing attacks
+      MessageDigest.isEqual(
+        providedCredentials.getBytes(StandardCharsets.UTF_8),
+        encodedCredentials.getBytes(StandardCharsets.UTF_8))
+    } catch {
+      case e: Exception =>
+        error("Error validating credentials", e)
+        false
+    }
+  }
+
+  private def sendUnauthorizedResponse(response: HttpServletResponse): Unit = {
+    response.setHeader("WWW-Authenticate", "Basic realm=\"Kyuubi Prometheus Metrics\"")
+    response.sendError(HttpServletResponse.SC_UNAUTHORIZED, "Unauthorized")
+  }
+}

kyuubi-metrics/src/main/scala/org/apache/kyuubi/metrics/MetricsConf.scala

@@ -19,7 +19,7 @@ package org.apache.kyuubi.metrics
 
 import java.time.Duration
 
-import org.apache.kyuubi.config.ConfigEntry
+import org.apache.kyuubi.config.{ConfigEntry, OptionalConfigEntry}
 import org.apache.kyuubi.config.KyuubiConf.buildConf
 import org.apache.kyuubi.metrics.ReporterType._
 
@@ -95,6 +95,27 @@ object MetricsConf {
       .booleanConf
       .createWithDefault(false)
 
+  val METRICS_PROMETHEUS_AUTH_ENABLED: ConfigEntry[Boolean] =
+    buildConf("kyuubi.metrics.prometheus.auth.enabled")
+      .doc("Enable basic authentication for Prometheus metrics endpoint")
+      .version("1.8.0")
+      .booleanConf
+      .createWithDefault(false)
+
+  val METRICS_PROMETHEUS_AUTH_USERNAME: OptionalConfigEntry[String] =
+    buildConf("kyuubi.metrics.prometheus.auth.username")
+      .doc("Username for Prometheus metrics endpoint basic authentication")
+      .version("1.8.0")
+      .stringConf
+      .createOptional
+
+  val METRICS_PROMETHEUS_AUTH_PASSWORD: OptionalConfigEntry[String] =
+    buildConf("kyuubi.metrics.prometheus.auth.password")
+      .doc("Password for Prometheus metrics endpoint basic authentication")
+      .version("1.8.0")
+      .stringConf
+      .createOptional
+
   val METRICS_SLF4J_INTERVAL: ConfigEntry[Long] = buildConf("kyuubi.metrics.slf4j.interval")
     .serverOnly
     .doc("How often should report metrics to SLF4J logger")

kyuubi-metrics/src/main/scala/org/apache/kyuubi/metrics/PrometheusReporterService.scala

@@ -17,6 +17,8 @@
 
 package org.apache.kyuubi.metrics
 
+import java.util
+import javax.servlet.DispatcherType
 import javax.servlet.http.{HttpServlet, HttpServletRequest, HttpServletResponse}
 
 import com.codahale.metrics.MetricRegistry
@@ -25,7 +27,7 @@ import io.prometheus.client.dropwizard.DropwizardExports
 import io.prometheus.client.exporter.MetricsServlet
 import io.prometheus.client.exporter.common.TextFormat
 import org.eclipse.jetty.server.{HttpConfiguration, HttpConnectionFactory, Server, ServerConnector}
-import org.eclipse.jetty.servlet.{ServletContextHandler, ServletHolder}
+import org.eclipse.jetty.servlet.{FilterHolder, ServletContextHandler, ServletHolder}
 
 import org.apache.kyuubi.KyuubiException
 import org.apache.kyuubi.config.KyuubiConf
@@ -59,6 +61,9 @@ class PrometheusReporterService(registry: MetricRegistry)
     context.setContextPath("/")
     httpServer.setHandler(context)
 
+    // Add BasicAuth filter if enabled
+    addBasicAuthFilterIfEnabled(conf, context)
+
     new DropwizardExports(registry).register(bridgeRegistry)
     if (conf.get(MetricsConf.METRICS_PROMETHEUS_LABELS_INSTANCE_ENABLED)) {
       val instanceLabel =
@@ -113,6 +118,44 @@ class PrometheusReporterService(registry: MetricRegistry)
     }
   }
 
+  /**
+   * Add BasicAuth filter to the servlet context if authentication is enabled
+   */
+  private def addBasicAuthFilterIfEnabled(
+      conf: KyuubiConf,
+      context: ServletContextHandler): Unit = {
+    val authEnabled = conf.get(MetricsConf.METRICS_PROMETHEUS_AUTH_ENABLED)
+
+    if (authEnabled) {
+      val username = conf.get(MetricsConf.METRICS_PROMETHEUS_AUTH_USERNAME)
+      val password = conf.get(MetricsConf.METRICS_PROMETHEUS_AUTH_PASSWORD)
+
+      (username, password) match {
+        case (Some(user), Some(pass)) =>
+          if (user.trim.isEmpty || pass.trim.isEmpty) {
+            throw new KyuubiException(
+              "Username and password cannot be empty when authentication is enabled")
+          }
+
+          val authFilter = new BasicAuthFilter(user, pass)
+          val filterHolder = new FilterHolder(authFilter)
+          context.addFilter(
+            filterHolder,
+            "/*",
+            util.EnumSet.of(DispatcherType.REQUEST))
+
+          info(s"BasicAuth enabled for Prometheus metrics endpoint with username: $user")
+
+        case _ =>
+          throw new KyuubiException(
+            "Both username and password must be configured when " +
+              "kyuubi.metrics.prometheus.auth.enabled is true")
+      }
+    } else {
+      info("BasicAuth disabled for Prometheus metrics endpoint")
+    }
+  }
+
   private def createPrometheusServletWithLabels(labels: Map[String, String]): HttpServlet = {
     new HttpServlet {
       override def doGet(request: HttpServletRequest, response: HttpServletResponse): Unit = {

kyuubi-metrics/src/test/scala/org/apache/kyuubi/metrics/MetricsSystemSuite.scala

@@ -71,6 +71,76 @@ class MetricsSystemSuite extends KyuubiFunSuite {
     metricsSystem.stop()
   }
 
+  test("metrics - PrometheusReporter With Authentication") {
+    val testContextPath = "/prometheus-metrics"
+
+    val conf = KyuubiConf()
+      .set(MetricsConf.METRICS_ENABLED, true)
+      .set(MetricsConf.METRICS_REPORTERS, Set(ReporterType.PROMETHEUS.toString))
+      .set(MetricsConf.METRICS_PROMETHEUS_PORT, 0) // random port
+      .set(MetricsConf.METRICS_PROMETHEUS_PATH, testContextPath)
+      .set(MetricsConf.METRICS_PROMETHEUS_AUTH_ENABLED, true)
+      .set(MetricsConf.METRICS_PROMETHEUS_AUTH_USERNAME, "admin")
+      .set(MetricsConf.METRICS_PROMETHEUS_AUTH_PASSWORD, "password")
+
+    val metricsSystem = new MetricsSystem()
+    metricsSystem.initialize(conf)
+    metricsSystem.start()
+
+    try {
+      metricsSystem.registerGauge(MetricsConstants.CONN_OPEN, 2021, 0)
+
+      val prometheusHttpServer = metricsSystem.getServices.head
+        .asInstanceOf[PrometheusReporterService].httpServer
+
+      val client: HttpClient = new HttpClient
+      client.start()
+
+      try {
+        // Test successful authentication with correct credentials
+        val request = client.newRequest(prometheusHttpServer.getURI.resolve(testContextPath))
+        val credentials = "admin:password"
+        val encodedCredentials = java.util.Base64.getEncoder.encodeToString(
+          credentials.getBytes(java.nio.charset.StandardCharsets.UTF_8))
+        request.header("Authorization", "Basic " + encodedCredentials)
+        val res: ContentResponse = request.send()
+
+        // Verify response contains expected metrics
+        assert(res.getStatus == 200, "Should return 200 OK with correct credentials")
+        assert(res.getContentAsString.contains("heap_usage"), "Should contain heap_usage metric")
+        assert(
+          res.getContentAsString.contains("kyuubi_connection_opened 2021.0"),
+          "Should contain registered gauge metric")
+
+        // Test authentication failure with wrong credentials
+        val wrongAuthRequest =
+          client.newRequest(prometheusHttpServer.getURI.resolve(testContextPath))
+        val wrongCredentials = "wrong:wrong"
+        val wrongEncodedCredentials = java.util.Base64.getEncoder.encodeToString(
+          wrongCredentials.getBytes(java.nio.charset.StandardCharsets.UTF_8))
+        wrongAuthRequest.header("Authorization", "Basic " + wrongEncodedCredentials)
+        val wrongAuthRes: ContentResponse = wrongAuthRequest.send()
+
+        // Should return 401 Unauthorized for wrong credentials
+        assert(
+          wrongAuthRes.getStatus == 401,
+          "Should return 401 Unauthorized with wrong credentials")
+
+        // Test authentication failure with no credentials
+        val noAuthRequest = client.newRequest(prometheusHttpServer.getURI.resolve(testContextPath))
+        val noAuthRes: ContentResponse = noAuthRequest.send()
+
+        // Should return 401 Unauthorized for missing credentials
+        assert(noAuthRes.getStatus == 401, "Should return 401 Unauthorized with no credentials")
+
+      } finally {
+        client.stop()
+      }
+    } finally {
+      metricsSystem.stop()
+    }
+  }
+
   test("metrics - other reporters") {
     val reportPath = Utils.createTempDir()
     val conf = KyuubiConf()