}:
+ *
+ * --conf spark.ui.filters=org.apache.spark.filter.MetricsAuthFilter
+ * # metrics token - OMIT this line for open metrics:
+ * --conf spark.org.apache.spark.filter.MetricsAuthFilter.param.token=<shared-secret>
+ * # optional SPNEGO for the rest of the UI (params carry a 'spnego.' prefix):
+ * --conf spark.org.apache.spark.filter.MetricsAuthFilter.param.spnego.type=kerberos
+ * --conf spark.org.apache.spark.filter.MetricsAuthFilter.param.spnego.kerberos.\
+ * principal=HTTP/_HOST@REALM
+ * --conf spark.org.apache.spark.filter.MetricsAuthFilter.param.spnego.kerberos.\
+ * keytab=/etc/keytabs/spnego.keytab
+ * # confidentiality: also enable TLS so the token is not sent in cleartext
+ * --conf spark.ssl.ui.enabled=true
+ *
+ * On the History Server the registration is identical, but TLS is
+ * {@code spark.ssl.historyServer.*} instead of {@code spark.ssl.ui.*}.
+ *
+ * Scraper side (vmagent / Prometheus)
+ *
+ * authorization: # only when a token is configured; drop it for open metrics
+ * type: Bearer
+ * credentials_file: /etc/scrape-auth/token
+ *
+ *
+ * NOTE: authentication only - not encryption. Pair with TLS so the bearer
+ * token and the metrics travel encrypted.
+ */
+public class MetricsAuthFilter implements Filter {
+
+ /** Path segment that identifies the Spark metrics endpoints. */
+ private static final String METRICS_PREFIX = "/metrics/";
+
+ /** Init-param prefix carrying the embedded AuthenticationFilter's config. */
+ private static final String SPNEGO_PREFIX = "spnego.";
+
+ /** null => metrics are OPEN (no token required). */
+ private byte[] expectedToken;
+
+ /** null => non-metrics paths fall through untouched (no SPNEGO). */
+ private Filter uiDelegate;
+
+ @Override
+ public void init(FilterConfig cfg) throws ServletException {
+ // Fed by spark..param.token - exactly what
+ // JettyUtils.addFilters reads via conf.getAllWithPrefix("spark.$filter.param.").
+ // Absent/empty token => open metrics.
+ String token = cfg.getInitParameter("token");
+ this.expectedToken = (token == null || token.isEmpty())
+ ? null
+ : token.getBytes(StandardCharsets.UTF_8);
+
+ // Optionally protect the rest of the UI with the built-in SPNEGO/pseudo
+ // filter. Enabled by presence of a 'spnego.type' param.
+ if (cfg.getInitParameter(SPNEGO_PREFIX + "type") != null) {
+ Filter delegate = createUiDelegate();
+ delegate.init(new PrefixedFilterConfig(cfg, SPNEGO_PREFIX));
+ this.uiDelegate = delegate;
+ }
+ }
+
+ /**
+ * The built-in filter that non-metrics UI paths are delegated to. On this
+ * (Spark 4.2.x) build it is the fork's jakarta {@link AuthenticationFilter}.
+ * Package-private and overridable so tests can inject a stub without Kerberos.
+ */
+ Filter createUiDelegate() {
+ return new AuthenticationFilter();
+ }
+
+ @Override
+ public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain)
+ throws IOException, ServletException {
+ HttpServletRequest req = (HttpServletRequest) request;
+ HttpServletResponse resp = (HttpServletResponse) response;
+
+ String uri = req.getRequestURI();
+ if (uri != null && uri.contains(METRICS_PREFIX)) {
+ // Metrics path: bearer token OR open (no-token mode). Never SPNEGO, so the
+ // scraper is not challenged with Kerberos.
+ if (expectedToken != null && !bearerMatches(req)) {
+ resp.setHeader("WWW-Authenticate", "Bearer");
+ resp.sendError(HttpServletResponse.SC_UNAUTHORIZED, "Invalid or missing bearer token");
+ return; // short-circuit
+ }
+ chain.doFilter(request, response); // straight to the metrics servlet
+ return;
+ }
+
+ // Non-metrics UI path: SPNEGO if configured, else fall through to Spark.
+ if (uiDelegate != null) {
+ uiDelegate.doFilter(request, response, chain); // authenticates, then continues chain
+ } else {
+ chain.doFilter(request, response);
+ }
+ }
+
+ private boolean bearerMatches(HttpServletRequest req) {
+ String auth = req.getHeader("Authorization");
+ if (auth == null || !auth.startsWith("Bearer ")) {
+ return false;
+ }
+ // Constant-time compare so the token cannot be recovered via response timing.
+ byte[] presented = auth.substring(7).getBytes(StandardCharsets.UTF_8);
+ return MessageDigest.isEqual(presented, expectedToken);
+ }
+
+ @Override
+ public void destroy() {
+ if (uiDelegate != null) {
+ uiDelegate.destroy();
+ uiDelegate = null;
+ }
+ }
+
+ /**
+ * A {@link FilterConfig} view that exposes only the init params starting with a
+ * given prefix, with the prefix stripped - so the embedded
+ * {@link AuthenticationFilter} sees {@code type}, {@code kerberos.principal},
+ * ... rather than {@code spnego.type}, {@code spnego.kerberos.principal}, and
+ * never sees our own {@code token}.
+ */
+ private static final class PrefixedFilterConfig implements FilterConfig {
+ private final FilterConfig delegate;
+ private final String prefix;
+
+ PrefixedFilterConfig(FilterConfig delegate, String prefix) {
+ this.delegate = delegate;
+ this.prefix = prefix;
+ }
+
+ @Override
+ public String getFilterName() {
+ return delegate.getFilterName();
+ }
+
+ @Override
+ public ServletContext getServletContext() {
+ return delegate.getServletContext();
+ }
+
+ @Override
+ public String getInitParameter(String name) {
+ return delegate.getInitParameter(prefix + name);
+ }
+
+ @Override
+ public Enumeration getInitParameterNames() {
+ List stripped = new ArrayList<>();
+ Enumeration all = delegate.getInitParameterNames();
+ while (all.hasMoreElements()) {
+ String name = all.nextElement();
+ if (name.startsWith(prefix)) {
+ stripped.add(name.substring(prefix.length()));
+ }
+ }
+ return Collections.enumeration(stripped);
+ }
+ }
+}
diff --git a/core/src/test/java/org/apache/spark/filter/MetricsAuthFilterSuite.java b/core/src/test/java/org/apache/spark/filter/MetricsAuthFilterSuite.java
new file mode 100644
index 0000000000000..0492bfa5ff55f
--- /dev/null
+++ b/core/src/test/java/org/apache/spark/filter/MetricsAuthFilterSuite.java
@@ -0,0 +1,295 @@
+/*
+ * 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.spark.filter;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.Enumeration;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import jakarta.servlet.Filter;
+import jakarta.servlet.FilterChain;
+import jakarta.servlet.FilterConfig;
+import jakarta.servlet.ServletContext;
+import jakarta.servlet.ServletException;
+import jakarta.servlet.ServletRequest;
+import jakarta.servlet.ServletResponse;
+import jakarta.servlet.http.HttpServletRequest;
+import jakarta.servlet.http.HttpServletResponse;
+
+import org.junit.jupiter.api.Test;
+
+/**
+ * Unit tests for {@link MetricsAuthFilter}. The SPNEGO delegate is replaced
+ * with a {@link RecordingFilter} via the package-private {@code createUiDelegate()}
+ * seam, so no Kerberos/Hadoop machinery is needed.
+ */
+class MetricsAuthFilterSuite {
+
+ private static final String TOKEN = "s3cr3t-shared-token";
+
+ // ---- /metrics/* : bearer / open -----------------------------------------
+
+ @Test
+ void metricsWithCorrectBearerPassesThrough() throws Exception {
+ RecordingFilter delegate = new RecordingFilter();
+ MetricsAuthFilter f = newFilter(params("token", TOKEN), delegate);
+
+ HttpServletRequest req = metricsRequest("Bearer " + TOKEN);
+ HttpServletResponse resp = mock(HttpServletResponse.class);
+ FilterChain chain = mock(FilterChain.class);
+
+ f.doFilter(req, resp, chain);
+
+ verify(chain, times(1)).doFilter(req, resp);
+ verify(resp, never()).sendError(org.mockito.ArgumentMatchers.anyInt(),
+ org.mockito.ArgumentMatchers.anyString());
+ assertEquals(0, delegate.doFilterCalls, "metrics path must never hit the UI delegate");
+ }
+
+ @Test
+ void metricsWithWrongTokenIsRejected() throws Exception {
+ MetricsAuthFilter f = newFilter(params("token", TOKEN), new RecordingFilter());
+
+ HttpServletRequest req = metricsRequest("Bearer not-the-token");
+ HttpServletResponse resp = mock(HttpServletResponse.class);
+ FilterChain chain = mock(FilterChain.class);
+
+ f.doFilter(req, resp, chain);
+
+ verify(resp).setHeader("WWW-Authenticate", "Bearer");
+ verify(resp).sendError(HttpServletResponse.SC_UNAUTHORIZED,
+ "Invalid or missing bearer token");
+ verify(chain, never()).doFilter(req, resp);
+ }
+
+ @Test
+ void metricsWithMissingHeaderIsRejected() throws Exception {
+ MetricsAuthFilter f = newFilter(params("token", TOKEN), new RecordingFilter());
+
+ HttpServletRequest req = metricsRequest(null); // no Authorization header
+ HttpServletResponse resp = mock(HttpServletResponse.class);
+ FilterChain chain = mock(FilterChain.class);
+
+ f.doFilter(req, resp, chain);
+
+ verify(resp).sendError(HttpServletResponse.SC_UNAUTHORIZED,
+ "Invalid or missing bearer token");
+ verify(chain, never()).doFilter(req, resp);
+ }
+
+ @Test
+ void metricsOpenWhenNoTokenConfigured() throws Exception {
+ MetricsAuthFilter f = newFilter(Collections.emptyMap(), new RecordingFilter());
+
+ HttpServletRequest req = metricsRequest(null); // no credentials at all
+ HttpServletResponse resp = mock(HttpServletResponse.class);
+ FilterChain chain = mock(FilterChain.class);
+
+ f.doFilter(req, resp, chain);
+
+ verify(chain, times(1)).doFilter(req, resp);
+ verify(resp, never()).sendError(org.mockito.ArgumentMatchers.anyInt(),
+ org.mockito.ArgumentMatchers.anyString());
+ }
+
+ @Test
+ void bearerOfDifferentLengthIsRejected() throws Exception {
+ // Guards the constant-time MessageDigest.isEqual path against length mismatch.
+ MetricsAuthFilter f = newFilter(params("token", TOKEN), new RecordingFilter());
+
+ HttpServletRequest req = metricsRequest("Bearer short");
+ HttpServletResponse resp = mock(HttpServletResponse.class);
+ FilterChain chain = mock(FilterChain.class);
+
+ f.doFilter(req, resp, chain);
+
+ verify(resp).sendError(HttpServletResponse.SC_UNAUTHORIZED,
+ "Invalid or missing bearer token");
+ verify(chain, never()).doFilter(req, resp);
+ }
+
+ // ---- non-metrics : fall-through vs SPNEGO delegate ----------------------
+
+ @Test
+ void nonMetricsFallsThroughWhenNoSpnego() throws Exception {
+ RecordingFilter delegate = new RecordingFilter();
+ MetricsAuthFilter f = newFilter(params("token", TOKEN), delegate);
+
+ HttpServletRequest req = uiRequest("/jobs/");
+ HttpServletResponse resp = mock(HttpServletResponse.class);
+ FilterChain chain = mock(FilterChain.class);
+
+ f.doFilter(req, resp, chain);
+
+ verify(chain, times(1)).doFilter(req, resp);
+ assertEquals(0, delegate.doFilterCalls, "delegate not created without spnego.type");
+ }
+
+ @Test
+ void nonMetricsIsDelegatedWhenSpnegoConfigured() throws Exception {
+ RecordingFilter delegate = new RecordingFilter();
+ Map p = params("token", TOKEN);
+ p.put("spnego.type", "simple");
+ MetricsAuthFilter f = newFilter(p, delegate);
+
+ HttpServletRequest req = uiRequest("/jobs/");
+ HttpServletResponse resp = mock(HttpServletResponse.class);
+ FilterChain chain = mock(FilterChain.class);
+
+ f.doFilter(req, resp, chain);
+
+ assertEquals(1, delegate.doFilterCalls, "UI path must reach the SPNEGO delegate");
+ verify(chain, times(1)).doFilter(req, resp); // RecordingFilter continues the chain
+ }
+
+ @Test
+ void metricsNeverReachesDelegateEvenWhenSpnegoConfigured() throws Exception {
+ RecordingFilter delegate = new RecordingFilter();
+ Map p = params("token", TOKEN);
+ p.put("spnego.type", "simple");
+ MetricsAuthFilter f = newFilter(p, delegate);
+
+ HttpServletRequest req = metricsRequest("Bearer " + TOKEN);
+ HttpServletResponse resp = mock(HttpServletResponse.class);
+ FilterChain chain = mock(FilterChain.class);
+
+ f.doFilter(req, resp, chain);
+
+ assertEquals(0, delegate.doFilterCalls, "scraper must never be challenged with SPNEGO");
+ verify(chain, times(1)).doFilter(req, resp);
+ }
+
+ // ---- PrefixedFilterConfig : prefix stripping + token hiding --------------
+
+ @Test
+ void delegateSeesStrippedParamsAndNotToken() throws Exception {
+ RecordingFilter delegate = new RecordingFilter();
+ Map p = params("token", TOKEN);
+ p.put("spnego.type", "kerberos");
+ p.put("spnego.kerberos.principal", "HTTP/_HOST@REALM");
+ MetricsAuthFilter f = newFilter(p, delegate);
+
+ FilterConfig seen = delegate.initConfig;
+ assertEquals("kerberos", seen.getInitParameter("type"), "prefix must be stripped");
+ assertEquals("HTTP/_HOST@REALM", seen.getInitParameter("kerberos.principal"));
+ assertNull(seen.getInitParameter("token"), "delegate must not see our own token");
+ assertNull(seen.getInitParameter("spnego.type"), "prefixed name must not leak");
+
+ List names = Collections.list(seen.getInitParameterNames());
+ assertTrue(names.contains("type"));
+ assertTrue(names.contains("kerberos.principal"));
+ assertFalse(names.contains("token"));
+ assertFalse(names.contains("spnego.type"));
+ }
+
+ // ---- helpers ------------------------------------------------------------
+
+ private static MetricsAuthFilter newFilter(Map initParams,
+ Filter delegate) throws ServletException {
+ MetricsAuthFilter f = new MetricsAuthFilter() {
+ @Override
+ Filter createUiDelegate() {
+ return delegate;
+ }
+ };
+ f.init(new MapFilterConfig(initParams));
+ return f;
+ }
+
+ private static HttpServletRequest metricsRequest(String authorization) {
+ HttpServletRequest req = mock(HttpServletRequest.class);
+ when(req.getRequestURI()).thenReturn("/metrics/prometheus/");
+ when(req.getHeader("Authorization")).thenReturn(authorization);
+ return req;
+ }
+
+ private static HttpServletRequest uiRequest(String uri) {
+ HttpServletRequest req = mock(HttpServletRequest.class);
+ when(req.getRequestURI()).thenReturn(uri);
+ return req;
+ }
+
+ private static Map params(String k, String v) {
+ Map m = new LinkedHashMap<>();
+ m.put(k, v);
+ return m;
+ }
+
+ /** A {@link Filter} that records how it was init'd and continues the chain. */
+ private static final class RecordingFilter implements Filter {
+ private FilterConfig initConfig;
+ private int doFilterCalls;
+
+ @Override
+ public void init(FilterConfig filterConfig) {
+ this.initConfig = filterConfig;
+ }
+
+ @Override
+ public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain)
+ throws IOException, ServletException {
+ doFilterCalls++;
+ chain.doFilter(request, response);
+ }
+
+ @Override
+ public void destroy() {
+ }
+ }
+
+ /** Minimal {@link FilterConfig} backed by a map, for driving {@code init}. */
+ private static final class MapFilterConfig implements FilterConfig {
+ private final Map params;
+
+ MapFilterConfig(Map params) {
+ this.params = params;
+ }
+
+ @Override
+ public String getFilterName() {
+ return "MetricsAuthFilter";
+ }
+
+ @Override
+ public ServletContext getServletContext() {
+ return null;
+ }
+
+ @Override
+ public String getInitParameter(String name) {
+ return params.get(name);
+ }
+
+ @Override
+ public Enumeration getInitParameterNames() {
+ return Collections.enumeration(params.keySet());
+ }
+ }
+}
diff --git a/docs/security.md b/docs/security.md
index b3743d9a1b0a6..17b371dbdfe08 100644
--- a/docs/security.md
+++ b/docs/security.md
@@ -322,8 +322,9 @@ The following settings cover enabling encryption for data written to disk:
## Authentication and Authorization
Enabling authentication for the Web UIs is done using [jakarta servlet filters](https://jakarta.ee/specifications/servlet/5.0/apidocs/jakarta/servlet/filter).
-You will need a filter that implements the authentication method you want to deploy. Spark does not
-provide any built-in authentication filters.
+You will need a filter that implements the authentication method you want to deploy. Apache Spark
+does not provide any built-in authentication filters, but this distribution ships the two described
+in [Built-in authentication filters](#built-in-authentication-filters) below.
Spark also supports access control to the UI when an authentication filter is present. Each
application can be configured with its own separate access control lists (ACLs). Spark
@@ -442,6 +443,47 @@ The following options control the authentication of Web UIs:
On YARN, the view and modify ACLs are provided to the YARN service when submitting applications, and
control who has the respective privileges via YARN interfaces.
+### Built-in authentication filters
+
+In addition to `org.apache.spark.ui.JWSFilter` (signed-JWT bearer auth), this distribution ships two
+servlet filters you can name in `spark.ui.filters`. Filter parameters are passed as
+`spark..param.` (the same convention Spark uses for any UI
+filter). Both ship in `spark-core`, so no extra jar is required. They authenticate only - pair them
+with UI TLS (`spark.ssl.ui.enabled=true`, or `spark.ssl.historyServer.*` on the History Server) so
+credentials are not sent in the clear.
+
+`org.apache.spark.filter.AuthenticationFilter` protects the **whole** Web UI with SPNEGO/Kerberos
+(`type=kerberos`) or pseudo (`type=simple`) authentication, backed by Hadoop's authentication
+handlers:
+
+```
+spark.ui.filters=org.apache.spark.filter.AuthenticationFilter
+spark.org.apache.spark.filter.AuthenticationFilter.param.type=kerberos
+spark.org.apache.spark.filter.AuthenticationFilter.param.kerberos.principal=HTTP/_HOST@REALM
+spark.org.apache.spark.filter.AuthenticationFilter.param.kerberos.keytab=/etc/security/keytabs/spnego.keytab
+```
+
+`org.apache.spark.filter.MetricsAuthFilter` applies **different** policies to the Prometheus metrics
+endpoints (`/metrics/*`) and to the rest of the UI - something plain configuration cannot express,
+because `spark.ui.filters`, the ACLs and the TLS connector are all UI-wide. It bearer-guards (or, if
+no `token` is set, leaves open) `/metrics/*` for a Prometheus/vmagent scraper that cannot perform
+Kerberos, while optionally delegating every other path to an embedded `AuthenticationFilter` for
+SPNEGO. Register **only** this filter (it embeds the SPNEGO delegate; do not also list
+`AuthenticationFilter`, which would challenge `/metrics/*` too):
+
+```
+spark.ui.filters=org.apache.spark.filter.MetricsAuthFilter
+# bearer token for /metrics/* (omit this line for open metrics):
+spark.org.apache.spark.filter.MetricsAuthFilter.param.token=
+# optional SPNEGO for the rest of the UI (params carry a 'spnego.' prefix):
+spark.org.apache.spark.filter.MetricsAuthFilter.param.spnego.type=kerberos
+spark.org.apache.spark.filter.MetricsAuthFilter.param.spnego.kerberos.principal=HTTP/_HOST@REALM
+spark.org.apache.spark.filter.MetricsAuthFilter.param.spnego.kerberos.keytab=/etc/security/keytabs/spnego.keytab
+```
+
+The scraper then presents the token as a `Bearer` credential; on the History Server the registration
+is identical (only the TLS namespace differs).
+
## Spark History Server ACLs
Authentication for the SHS Web UI is enabled the same way as for regular applications, using