supportedSchemes();
-
- /**
- * Exchanges the user's identity for a short-lived service credential scoped to the
- * given target URI.
- *
- * For example, an AWS implementation might call STS AssumeRoleWithWebIdentity using
- * the raw token from the {@link UserContext} and return temporary AWS credentials as
- * a {@link ServiceCredential}.
- *
- * @param user the authenticated user context containing the identity token (must not be null)
- * @param target the target URI for which credentials are requested (must not be null)
- * @return a short-lived service credential for the target
- * @throws CredentialResolutionException if the credential exchange fails
- * @since 4.3.0
- */
- ServiceCredential resolve(UserContext user, URI target) throws CredentialResolutionException;
-
- /**
- * Returns the suggested time-to-live for credentials produced by this provider.
- *
- * The credential management layer uses this as a hint for refresh scheduling.
- * The default is 15 minutes.
- *
- * @return the suggested credential TTL (never null)
- * @since 4.3.0
- */
- default Duration suggestedTtl() {
- return Duration.ofMinutes(15);
- }
-
- /**
- * Releases any resources held by this provider (e.g., HTTP clients, connection pools).
- *
- * Called by the credential management layer during shutdown. The default implementation
- * is a no-op; providers that allocate long-lived resources in {@link #init(Map)} should
- * override this method to clean them up.
- *
- * {@code close()} may be invoked while another thread is still executing
- * {@link #resolve(UserContext, URI)}: shutdown interrupts the renewal thread but does
- * not wait for in-flight calls to complete. Implementations must tolerate a concurrent
- * or subsequent {@code resolve()} failing after resources have been released, and
- * {@code close()} itself must not block indefinitely.
- *
- * Implementations that do not throw checked exceptions may narrow the {@code throws}
- * clause in their override (e.g., declare {@code close()} with no {@code throws} or
- * with a more specific exception type).
- *
- * @since 4.3.0
- */
- @Override
- default void close() throws Exception {}
-}
diff --git a/core/src/main/java/org/apache/spark/security/CredentialProviderLoader.java b/core/src/main/java/org/apache/spark/security/CredentialProviderLoader.java
deleted file mode 100644
index 84aaaca0a1961..0000000000000
--- a/core/src/main/java/org/apache/spark/security/CredentialProviderLoader.java
+++ /dev/null
@@ -1,321 +0,0 @@
-/*
- * 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.security;
-
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.IdentityHashMap;
-import java.util.List;
-import java.util.Locale;
-import java.util.Map;
-import java.util.Objects;
-import java.util.Optional;
-import java.util.ServiceLoader;
-import java.util.Set;
-import java.util.stream.Collectors;
-
-import com.google.common.annotations.VisibleForTesting;
-
-import org.apache.spark.annotation.Private;
-
-/**
- * :: Private ::
- * Discovers {@link CredentialProvider} implementations via {@link ServiceLoader} and selects
- * the appropriate provider for a given URI scheme using Binding Policy A (explicit selection).
- *
- * Provider discovery happens once (lazily on first call) and the list is cached. Each provider
- * is initialized exactly once per provider instance via {@link CredentialProvider#init(Map)}
- * with the configuration from the first call that selects it (first-conf-wins semantics);
- * subsequent resolutions reuse the already-initialized instance without re-calling {@code init}.
- *
- * When the configuration key {@code spark.security.oidc.provider.} is set
- * (non-empty), the loader validates the configured fully-qualified class name against the
- * discovered candidates for that scheme regardless of candidate count. If the configured class
- * does not match any candidate, an {@link IllegalArgumentException} is thrown listing the
- * scheme, the configured class, and the available candidates. Only when the configuration key
- * is unset (or empty) do the count-based rules apply: a single candidate is auto-selected;
- * multiple candidates produce an ambiguity error; no candidates produce {@code Optional.empty()}.
- *
- * Thread-safety: This class uses synchronized access to the cached provider list and
- * initialization tracking, and callers may invoke {@link #providerFor(String, Map)} from
- * multiple threads. A provider instance is cached and shared across callers; per the
- * {@link CredentialProvider} contract, implementations must be thread-safe, so a returned
- * instance may be used concurrently.
- *
- * @since 4.3.0
- */
-@Private
-public final class CredentialProviderLoader {
-
- /**
- * Configuration key prefix for explicit provider selection per scheme.
- * When set (non-empty), the configured fully-qualified class name must match a discovered
- * provider that supports the scheme; a mismatch results in an {@link IllegalArgumentException}.
- */
- private static final String CONF_PREFIX = "spark.security.oidc.provider.";
-
- /**
- * Configuration key prefix used to scope the configuration passed to
- * {@link CredentialProvider#init(Map)}. Only keys starting with this prefix are forwarded,
- * preventing unrelated secrets from leaking to third-party provider implementations.
- */
- private static final String OIDC_CONF_PREFIX = "spark.security.oidc.";
-
- private static volatile List cachedProviders;
-
- /**
- * Tracks which provider instances have already been initialized. Guarded by the class lock.
- * Uses identity semantics (reference equality) to handle multiple provider instances correctly.
- */
- private static final Set initializedProviders =
- Collections.newSetFromMap(new IdentityHashMap<>());
-
- private CredentialProviderLoader() {
- // utility class
- }
-
- /**
- * Returns the {@link CredentialProvider} for the given URI scheme, applying Binding Policy A:
- *
- * - If no candidate supports the scheme, {@link Optional#empty()} is returned.
- * - If {@code spark.security.oidc.provider.} is set (non-empty) in
- * {@code conf}, the provider whose fully-qualified class name matches is selected
- * regardless of candidate count. If the configured class does not match any candidate,
- * an {@link IllegalArgumentException} is thrown naming the scheme, the configured class,
- * and the available candidates.
- * - If unset and exactly one candidate supports the scheme, that candidate is selected.
- * - If unset and multiple candidates support the scheme, an
- * {@link IllegalArgumentException} is thrown listing the candidates.
- *
- * The selected provider is initialized exactly once per provider instance via
- * {@link CredentialProvider#init(Map)} (first-conf-wins semantics); later resolutions reuse
- * the initialized instance without re-calling {@code init}. The configuration passed to
- * {@code init()} is scoped to {@code spark.security.oidc.*} keys only; keys from
- * other subsystems are filtered out to prevent secret leakage to third-party providers.
- *
- * Spark-internal callers pass their configuration as a {@code Map} for
- * testability and to match the signature of {@link CredentialProvider#init(Map)}.
- *
- * @param scheme the URI scheme (e.g., "s3a"); normalized to lowercase
- * @param conf Spark configuration properties as a string map
- * @return the selected provider, or empty if no provider supports the scheme
- * @throws IllegalArgumentException if explicit selection names an unknown or non-supporting
- * class, or if multiple candidates exist without explicit selection
- * @throws IllegalStateException if a provider returns null from {@code supportedSchemes()}
- */
- public static Optional providerFor(String scheme, Map conf) {
- Objects.requireNonNull(scheme, "scheme must not be null");
- Objects.requireNonNull(conf, "conf must not be null");
- if (scheme.isEmpty()) {
- throw new IllegalArgumentException("scheme must not be empty");
- }
- String normalizedScheme = scheme.toLowerCase(Locale.ROOT);
- List providers = getProviders();
-
- List candidates = providers.stream()
- .filter(p -> {
- Set schemes = p.supportedSchemes();
- if (schemes == null) {
- throw new IllegalStateException(
- "Provider " + p.getClass().getName()
- + " returned null from supportedSchemes()");
- }
- return schemes.stream()
- .anyMatch(s -> s.toLowerCase(Locale.ROOT).equals(normalizedScheme));
- })
- .collect(Collectors.toList());
-
- if (candidates.isEmpty()) {
- return Optional.empty();
- }
-
- String confKey = CONF_PREFIX + normalizedScheme;
- String explicitClass = conf.get(confKey);
-
- CredentialProvider selected;
- if (explicitClass != null && !explicitClass.isEmpty()) {
- selected = candidates.stream()
- .filter(p -> p.getClass().getName().equals(explicitClass))
- .findFirst()
- .orElseThrow(() -> new IllegalArgumentException(
- "Configured credential provider class '" + explicitClass + "' for scheme '"
- + normalizedScheme + "' (key: " + confKey
- + ") was not found among candidates or does not support the scheme. "
- + "Available candidates: "
- + candidates.stream()
- .map(p -> p.getClass().getName())
- .collect(Collectors.joining(", "))));
- } else if (candidates.size() == 1) {
- selected = candidates.get(0);
- } else {
- String candidateNames = candidates.stream()
- .map(p -> p.getClass().getName())
- .collect(Collectors.joining(", "));
- throw new IllegalArgumentException(
- "Multiple credential providers found for scheme '" + normalizedScheme
- + "'. Set " + confKey + " to one of: " + candidateNames);
- }
-
- // Initialize exactly once under the lock (first-conf-wins).
- // Only pass spark.security.oidc.* keys to init() to avoid leaking secrets
- // from other subsystems to third-party ServiceLoader providers. This follows the
- // precedent of DataSourceV2Utils.extractSessionConfigs() which scopes configuration
- // to a specific prefix. We keep the full key (unlike extractSessionConfigs which
- // strips the prefix) so providers can distinguish sub-keys unambiguously.
- synchronized (CredentialProviderLoader.class) {
- if (!initializedProviders.contains(selected)) {
- Map filteredConf = new HashMap<>();
- for (Map.Entry entry : conf.entrySet()) {
- if (entry.getKey().startsWith(OIDC_CONF_PREFIX)) {
- filteredConf.put(entry.getKey(), entry.getValue());
- }
- }
- selected.init(Collections.unmodifiableMap(filteredConf));
- initializedProviders.add(selected);
- }
- }
- return Optional.of(selected);
- }
-
- /**
- * Returns the cached list of discovered providers, loading them on first access.
- */
- private static List getProviders() {
- List providers = cachedProviders;
- if (providers == null) {
- synchronized (CredentialProviderLoader.class) {
- providers = cachedProviders;
- if (providers == null) {
- providers = loadProviders();
- cachedProviders = providers;
- }
- }
- }
- return providers;
- }
-
- private static List loadProviders() {
- ClassLoader cl = Thread.currentThread().getContextClassLoader();
- if (cl == null) {
- cl = CredentialProvider.class.getClassLoader();
- }
- ServiceLoader loader = ServiceLoader.load(CredentialProvider.class, cl);
- List result = new ArrayList<>();
- for (CredentialProvider provider : loader) {
- result.add(provider);
- }
- return result;
- }
-
- /**
- * Discovers all URI schemes supported by providers on the classpath.
- *
- * This method queries all discovered {@link CredentialProvider} instances and collects
- * their {@link CredentialProvider#supportedSchemes()} into a single set. Unlike
- * {@link #providerFor(String, Map)}, this does not initialize providers or apply
- * explicit selection rules -- it only reports what schemes are potentially available.
- *
- * Intended for use by {@code UserCredentialManager} when no explicit scheme configuration
- * (e.g., {@code spark.security.oidc.provider.}) is provided.
- *
- * @return a set of all supported scheme names (lowercased), possibly empty
- */
- public static Set discoverAllSchemes() {
- List providers = getProviders();
- Set schemes = new HashSet<>();
- for (CredentialProvider provider : providers) {
- Set providerSchemes = provider.supportedSchemes();
- if (providerSchemes != null) {
- for (String s : providerSchemes) {
- schemes.add(s.toLowerCase(Locale.ROOT));
- }
- }
- }
- return schemes;
- }
-
- /**
- * Closes all initialized providers, suppressing individual close exceptions.
- *
- * This method iterates over all providers that have been initialized via
- * {@link CredentialProvider#init(Map)} and calls {@link CredentialProvider#close()}
- * on each. If any provider's {@code close()} throws, the exception is suppressed
- * and attached to the first exception encountered. If at least one exception occurred,
- * it is thrown after all providers have been attempted.
- *
- * After this method returns (normally or exceptionally), the initialization tracking
- * is cleared, but the cached provider list is retained. This means providers would be
- * re-initialized on the next {@link #providerFor} call (which is not expected after
- * shutdown).
- *
- * Contract: {@code close()} implementations must not call back into
- * {@code CredentialProviderLoader} methods (e.g., {@code providerFor}).
- *
- * @throws Exception if one or more providers threw during close
- */
- public static void closeAll() throws Exception {
- List toClose;
- synchronized (CredentialProviderLoader.class) {
- // Copy and clear under the lock to prevent double-close if closeAll() is called
- // again concurrently, and to avoid ConcurrentModificationException.
- toClose = new ArrayList<>(initializedProviders);
- initializedProviders.clear();
- }
- // Close outside the lock so a slow or blocking close() cannot stall
- // providerFor() callers or deadlock against them.
- Exception firstException = null;
- for (CredentialProvider provider : toClose) {
- try {
- provider.close();
- } catch (Exception e) {
- if (firstException == null) {
- firstException = e;
- } else {
- firstException.addSuppressed(e);
- }
- }
- }
- if (firstException != null) {
- throw firstException;
- }
- }
-
- /**
- * Resets the cached provider list and initialization tracking. Intended for testing only.
- */
- @VisibleForTesting
- public static void resetForTesting() {
- synchronized (CredentialProviderLoader.class) {
- cachedProviders = null;
- initializedProviders.clear();
- }
- }
-
- /**
- * Overrides the cached provider list for testing. Intended for testing only.
- */
- @VisibleForTesting
- static void setProvidersForTesting(List providers) {
- synchronized (CredentialProviderLoader.class) {
- cachedProviders = providers;
- initializedProviders.clear();
- }
- }
-}
diff --git a/core/src/main/java/org/apache/spark/security/CredentialResolutionException.java b/core/src/main/java/org/apache/spark/security/CredentialResolutionException.java
deleted file mode 100644
index eb5aaf2372984..0000000000000
--- a/core/src/main/java/org/apache/spark/security/CredentialResolutionException.java
+++ /dev/null
@@ -1,54 +0,0 @@
-/*
- * 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.security;
-
-import org.apache.spark.annotation.DeveloperApi;
-
-/**
- * :: DeveloperApi ::
- * Thrown when a {@link CredentialProvider} fails to resolve credentials for a target URI.
- *
- * This is a checked exception to ensure callers handle credential resolution failures
- * explicitly (e.g., retry, fail the job, or fall back to another mechanism).
- *
- * @since 4.3.0
- */
-@DeveloperApi
-public class CredentialResolutionException extends Exception {
-
- private static final long serialVersionUID = 1L;
-
- /**
- * Constructs a new exception with the specified detail message.
- *
- * @param message the detail message
- */
- public CredentialResolutionException(String message) {
- super(message);
- }
-
- /**
- * Constructs a new exception with the specified detail message and cause.
- *
- * @param message the detail message
- * @param cause the underlying cause
- */
- public CredentialResolutionException(String message, Throwable cause) {
- super(message, cause);
- }
-}
diff --git a/core/src/main/java/org/apache/spark/security/FileTokenIngestor.java b/core/src/main/java/org/apache/spark/security/FileTokenIngestor.java
deleted file mode 100644
index b51dcc7f0b43e..0000000000000
--- a/core/src/main/java/org/apache/spark/security/FileTokenIngestor.java
+++ /dev/null
@@ -1,159 +0,0 @@
-/*
- * 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.security;
-
-import java.nio.charset.StandardCharsets;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.time.Instant;
-import java.util.Base64;
-import java.util.Optional;
-
-import com.fasterxml.jackson.databind.JsonNode;
-import com.fasterxml.jackson.databind.ObjectMapper;
-
-import org.apache.spark.annotation.Private;
-import org.apache.spark.internal.LogKeys;
-import org.apache.spark.internal.MDC;
-import org.apache.spark.internal.SparkLogger;
-import org.apache.spark.internal.SparkLoggerFactory;
-
-/**
- * A {@link TokenIngestor} that reads an OIDC identity token from a file.
- *
- * The file path typically points to a Kubernetes projected service account token
- * (e.g., {@code /var/run/secrets/tokens/spark-identity}) or a path configured via
- * {@code spark.security.oidc.identityToken.file}.
- *
- * This implementation:
- *
- * - Detects file rotation by comparing file content (only re-parses when it changes)
- * - Parses JWT claims by Base64-decoding the payload segment directly, without
- * signature verification, since the token is trusted from the local filesystem
- * and works with both signed (RS256, ES256) and unsigned tokens
- * - Handles errors gracefully: malformed JWT, missing file, or empty content
- * returns empty rather than throwing
- *
- *
- * @since 4.3.0
- */
-@Private
-public class FileTokenIngestor implements TokenIngestor {
-
- private static final SparkLogger LOG = SparkLoggerFactory.getLogger(FileTokenIngestor.class);
- private static final ObjectMapper MAPPER = new ObjectMapper();
-
- private final Path tokenPath;
-
- private volatile CachedToken cachedToken;
-
- private static final class CachedToken {
- private final String content;
- private final UserContext context;
-
- private CachedToken(String content, UserContext context) {
- this.content = content;
- this.context = context;
- }
- }
-
- /**
- * Construct a new FileTokenIngestor.
- *
- * @param tokenPath path to the identity token file (must not be null)
- */
- public FileTokenIngestor(Path tokenPath) {
- this.tokenPath = tokenPath;
- }
-
- @Override
- public Optional load() {
- try {
- if (!Files.exists(tokenPath)) {
- LOG.debug("Token file does not exist: {}", tokenPath);
- return Optional.empty();
- }
-
- String content = new String(Files.readAllBytes(tokenPath), StandardCharsets.UTF_8).trim();
- if (content.isEmpty()) {
- LOG.warn("Token file is empty: {}", MDC.of(LogKeys.PATH, tokenPath));
- return Optional.empty();
- }
-
- CachedToken currentCache = cachedToken;
- if (currentCache != null && content.equals(currentCache.content)) {
- return Optional.of(currentCache.context);
- }
-
- Optional userContext = parseJwt(content);
- if (userContext.isPresent()) {
- // If the new file has invalid content, return empty and do NOT
- // fall back to the previously cached context. The caller will retry on next poll.
- cachedToken = new CachedToken(content, userContext.get());
- }
- return userContext;
- } catch (Exception e) {
- LOG.warn("Failed to load token from {}: {}", e,
- MDC.of(LogKeys.PATH, tokenPath),
- MDC.of(LogKeys.CLASS_NAME, e.getClass().getName()));
- return Optional.empty();
- }
- }
-
- /**
- * Parse a JWT token string into a UserContext by Base64-decoding the payload segment.
- * This works with both signed (RS256, ES256) and unsigned (alg:none) tokens since
- * we never verify the signature - the token is trusted from the local filesystem.
- */
- private Optional parseJwt(String token) {
- try {
- String[] parts = token.split("\\.");
- if (parts.length < 2) {
- LOG.warn("JWT token does not have the expected header.payload format");
- return Optional.empty();
- }
-
- // Decode the payload (second segment) - works for both signed and unsigned JWTs
- byte[] payloadBytes = Base64.getUrlDecoder().decode(parts[1]);
- JsonNode claims = MAPPER.readTree(payloadBytes);
-
- String subject = claims.has("sub") ? claims.get("sub").asText() : null;
- String issuer = claims.has("iss") ? claims.get("iss").asText() : null;
-
- if (subject == null || subject.isEmpty()) {
- LOG.warn("JWT token missing required 'sub' claim");
- return Optional.empty();
- }
- if (issuer == null || issuer.isEmpty()) {
- LOG.warn("JWT token missing required 'iss' claim");
- return Optional.empty();
- }
-
- Instant issuedAt = claims.has("iat")
- ? Instant.ofEpochSecond(claims.get("iat").asLong()) : null;
- Instant expiresAt = claims.has("exp")
- ? Instant.ofEpochSecond(claims.get("exp").asLong()) : null;
-
- return Optional.of(new UserContext(subject, issuer, token, issuedAt, expiresAt));
- } catch (Exception e) {
- LOG.warn("Failed to parse JWT token: {}",
- MDC.of(LogKeys.CLASS_NAME, e.getClass().getName()));
- return Optional.empty();
- }
- }
-}
diff --git a/core/src/main/java/org/apache/spark/security/ServiceCredential.java b/core/src/main/java/org/apache/spark/security/ServiceCredential.java
deleted file mode 100644
index b0502cf922a45..0000000000000
--- a/core/src/main/java/org/apache/spark/security/ServiceCredential.java
+++ /dev/null
@@ -1,122 +0,0 @@
-/*
- * 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.security;
-
-import java.io.Serializable;
-import java.time.Instant;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Objects;
-
-import org.apache.spark.annotation.DeveloperApi;
-
-/**
- * :: DeveloperApi ::
- * A short-lived, service-specific credential derived from the user's identity token.
- *
- * Instances are produced by credential providers on the driver and transmitted
- * to executors via {@link UserCredentials}. The {@code properties} map holds
- * service-specific key-value pairs (e.g., temporary AWS credentials).
- *
- * This class is immutable and {@link Serializable}.
- *
- * @since 4.3.0
- */
-@DeveloperApi
-public final class ServiceCredential implements Serializable {
-
- private static final long serialVersionUID = 1L;
-
- private final Map properties;
- private final Instant expiresAt;
-
- /**
- * Constructs a new {@code ServiceCredential}.
- *
- * @param properties service-specific credential properties (must not be null; defensively copied)
- * @param expiresAt credential expiry time (may be null)
- */
- public ServiceCredential(Map properties, Instant expiresAt) {
- Objects.requireNonNull(properties, "properties must not be null");
- this.properties = new HashMap<>(properties);
- this.expiresAt = expiresAt;
- }
-
- /**
- * Returns an unmodifiable view of the credential properties.
- */
- public Map getProperties() {
- return Collections.unmodifiableMap(properties);
- }
-
- /** Returns the credential expiry time, or {@code null} if not set. */
- public Instant getExpiresAt() {
- return expiresAt;
- }
-
- /**
- * Returns {@code true} if this credential has expired relative to the given instant.
- * If {@code expiresAt} is {@code null}, this method returns {@code false}.
- *
- * @param now the current time to compare against (must not be null)
- * @return whether the credential is expired
- */
- public boolean isExpired(Instant now) {
- Objects.requireNonNull(now, "now must not be null");
- return expiresAt != null && !now.isBefore(expiresAt);
- }
-
- @Override
- public String toString() {
- String redactedProps;
- if (properties.isEmpty()) {
- redactedProps = "{}";
- } else {
- StringBuilder sb = new StringBuilder("{");
- boolean first = true;
- for (String key : properties.keySet()) {
- if (!first) {
- sb.append(", ");
- }
- sb.append(key).append("=[REDACTED]");
- first = false;
- }
- sb.append("}");
- redactedProps = sb.toString();
- }
- return "ServiceCredential{" +
- "properties=" + redactedProps +
- ", expiresAt=" + expiresAt +
- '}';
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) return true;
- if (o == null || getClass() != o.getClass()) return false;
- ServiceCredential that = (ServiceCredential) o;
- return properties.equals(that.properties)
- && Objects.equals(expiresAt, that.expiresAt);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(properties, expiresAt);
- }
-}
diff --git a/core/src/main/java/org/apache/spark/security/TokenIngestor.java b/core/src/main/java/org/apache/spark/security/TokenIngestor.java
deleted file mode 100644
index 0c25fc59d69e2..0000000000000
--- a/core/src/main/java/org/apache/spark/security/TokenIngestor.java
+++ /dev/null
@@ -1,46 +0,0 @@
-/*
- * 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.security;
-
-import java.util.Optional;
-
-import org.apache.spark.annotation.DeveloperApi;
-
-/**
- * :: DeveloperApi ::
- * Reads an OIDC identity token and produces a {@link UserContext}.
- *
- * Implementation should be stateless with respect to Spark configuration;
- * configuration is passed at construction time.
- * Implementations must be thread-safe because {@link #load()} may be called concurrently.
- *
- * @since 4.3.0
- */
-@DeveloperApi
-public interface TokenIngestor {
-
- /**
- * Attempt to load the current identity token and parse it into a UserContext.
- * This method may be called repeatedly. Implementations may cache parsed tokens, but must
- * detect changes to the underlying token source and return the current identity.
- *
- * @return a present Optional containing the UserContext if a valid token is available,
- * or empty if unavailable (e.g. empty content / missing file).
- */
- Optional load();
-}
diff --git a/core/src/main/java/org/apache/spark/security/UserContext.java b/core/src/main/java/org/apache/spark/security/UserContext.java
deleted file mode 100644
index 9d9f372dad34b..0000000000000
--- a/core/src/main/java/org/apache/spark/security/UserContext.java
+++ /dev/null
@@ -1,139 +0,0 @@
-/*
- * 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.security;
-
-import java.time.Instant;
-import java.util.Objects;
-
-import org.apache.spark.annotation.DeveloperApi;
-
-/**
- * :: DeveloperApi ::
- * Represents the authenticated user's identity context on the driver side.
- *
- * This class holds the OIDC token information used to derive short-lived
- * {@link ServiceCredential} instances via credential providers. It is intentionally
- * not {@link java.io.Serializable} and must never be transmitted to executors.
- * The {@code rawToken} field is always redacted in {@link #toString()}.
- *
- * @since 4.3.0
- */
-@DeveloperApi
-public final class UserContext {
-
- private final String principal;
- private final String issuer;
- private final String rawToken;
- private final Instant issuedAt;
- private final Instant expiresAt;
-
- /**
- * Constructs a new {@code UserContext}.
- *
- * @param principal the {@code sub} claim from the JWT (must not be null)
- * @param issuer the {@code iss} claim from the JWT (must not be null)
- * @param rawToken the raw OIDC JWT string (must not be null)
- * @param issuedAt token issue time (may be null)
- * @param expiresAt token expiry time (may be null)
- */
- public UserContext(
- String principal,
- String issuer,
- String rawToken,
- Instant issuedAt,
- Instant expiresAt) {
- this.principal = Objects.requireNonNull(principal, "principal must not be null");
- this.issuer = Objects.requireNonNull(issuer, "issuer must not be null");
- this.rawToken = Objects.requireNonNull(rawToken, "rawToken must not be null");
- this.issuedAt = issuedAt;
- this.expiresAt = expiresAt;
- }
-
- /** Returns the {@code sub} claim (principal identifier). */
- public String getPrincipal() {
- return principal;
- }
-
- /** Returns the {@code iss} claim (token issuer). */
- public String getIssuer() {
- return issuer;
- }
-
- /** Returns the raw OIDC JWT. This value must never be logged or transmitted to executors. */
- public String getRawToken() {
- return rawToken;
- }
-
- /** Returns the token issue time, or {@code null} if not set. */
- public Instant getIssuedAt() {
- return issuedAt;
- }
-
- /** Returns the token expiry time, or {@code null} if not set. */
- public Instant getExpiresAt() {
- return expiresAt;
- }
-
- /**
- * Returns {@code true} if this context's token has expired relative to the given instant.
- * If {@code expiresAt} is {@code null}, this method returns {@code false}.
- *
- * @param now the current time to compare against (must not be null)
- * @return whether the token is expired
- */
- public boolean isExpired(Instant now) {
- Objects.requireNonNull(now, "now must not be null");
- return expiresAt != null && !now.isBefore(expiresAt);
- }
-
- /**
- * Returns a string representation with the {@code rawToken} redacted as {@code [REDACTED]}.
- */
- @Override
- public String toString() {
- return "UserContext{" +
- "principal='" + principal + '\'' +
- ", issuer='" + issuer + '\'' +
- ", rawToken='[REDACTED]'" +
- ", issuedAt=" + issuedAt +
- ", expiresAt=" + expiresAt +
- '}';
- }
-
- /**
- * Equality is based on identity fields ({@code principal}, {@code issuer}) and token
- * validity window ({@code issuedAt}, {@code expiresAt}). The secret {@code rawToken} is
- * intentionally excluded: {@code UserContext} is driver-only and never used as a map key,
- * so there is no need to compare secret material.
- */
- @Override
- public boolean equals(Object o) {
- if (this == o) return true;
- if (o == null || getClass() != o.getClass()) return false;
- UserContext that = (UserContext) o;
- return principal.equals(that.principal)
- && issuer.equals(that.issuer)
- && Objects.equals(issuedAt, that.issuedAt)
- && Objects.equals(expiresAt, that.expiresAt);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(principal, issuer, issuedAt, expiresAt);
- }
-}
diff --git a/core/src/main/java/org/apache/spark/security/UserCredentials.java b/core/src/main/java/org/apache/spark/security/UserCredentials.java
deleted file mode 100644
index 3f5e1d54b9239..0000000000000
--- a/core/src/main/java/org/apache/spark/security/UserCredentials.java
+++ /dev/null
@@ -1,113 +0,0 @@
-/*
- * 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.security;
-
-import java.io.Serializable;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Locale;
-import java.util.Map;
-import java.util.Objects;
-import java.util.Optional;
-
-import org.apache.spark.annotation.DeveloperApi;
-
-/**
- * :: DeveloperApi ::
- * A bundle of {@link ServiceCredential} instances keyed by scheme (e.g., "s3a", "abfss").
- *
- * Scheme keys are normalized to lowercase ({@link Locale#ROOT}) at construction time, and
- * lookups via {@link #forScheme(String)} are case-insensitive. If the supplied map contains
- * keys that differ only by case, an {@link IllegalArgumentException} is thrown.
- *
- * This class is transmitted to executors and does not contain any reference
- * to {@link UserContext} or raw identity tokens. It is immutable and {@link Serializable}.
- *
- * @since 4.3.0
- */
-@DeveloperApi
-public final class UserCredentials implements Serializable {
-
- private static final long serialVersionUID = 1L;
-
- private final Map credentials;
-
- /**
- * Constructs a new {@code UserCredentials} bundle.
- *
- * Scheme keys are normalized to lowercase using {@link Locale#ROOT}. If multiple keys
- * collide after lowercasing (e.g. {@code "s3a"} and {@code "S3A"}), an
- * {@link IllegalArgumentException} is thrown rather than silently keeping one of them.
- *
- * @param credentials per-scheme map of service credentials (must not be null; defensively copied)
- * @throws IllegalArgumentException if two keys collide after case normalization
- */
- public UserCredentials(Map credentials) {
- Objects.requireNonNull(credentials, "credentials must not be null");
- Map normalized = new HashMap<>(credentials.size());
- for (Map.Entry entry : credentials.entrySet()) {
- Objects.requireNonNull(entry.getKey(), "scheme key must not be null");
- String scheme = entry.getKey().toLowerCase(Locale.ROOT);
- if (normalized.containsKey(scheme)) {
- throw new IllegalArgumentException(
- "Duplicate scheme after case normalization: " + entry.getKey());
- }
- normalized.put(scheme, entry.getValue());
- }
- this.credentials = normalized;
- }
-
- /**
- * Returns an unmodifiable view of all credentials keyed by scheme.
- */
- public Map getCredentials() {
- return Collections.unmodifiableMap(credentials);
- }
-
- /**
- * Looks up the {@link ServiceCredential} for the given scheme. The lookup is
- * case-insensitive (the argument is lowercased with {@link Locale#ROOT}).
- *
- * @param scheme the target scheme (e.g., "s3a", "S3A"); must not be null
- * @return an {@link Optional} containing the credential, or empty if no credential is registered
- */
- public Optional forScheme(String scheme) {
- Objects.requireNonNull(scheme, "scheme must not be null");
- return Optional.ofNullable(credentials.get(scheme.toLowerCase(Locale.ROOT)));
- }
-
- @Override
- public String toString() {
- return "UserCredentials{" +
- "credentials=" + credentials +
- '}';
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) return true;
- if (o == null || getClass() != o.getClass()) return false;
- UserCredentials that = (UserCredentials) o;
- return credentials.equals(that.credentials);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(credentials);
- }
-}
diff --git a/core/src/main/scala/org/apache/spark/deploy/security/UserCredentialManager.scala b/core/src/main/scala/org/apache/spark/deploy/security/UserCredentialManager.scala
deleted file mode 100644
index 0627a87d9436e..0000000000000
--- a/core/src/main/scala/org/apache/spark/deploy/security/UserCredentialManager.scala
+++ /dev/null
@@ -1,461 +0,0 @@
-/*
- * 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.deploy.security
-
-import java.io.{ByteArrayOutputStream, ObjectInputFilter, ObjectInputStream, ObjectOutputStream}
-import java.net.URI
-import java.nio.file.Paths
-import java.time.Instant
-import java.util.concurrent.{RejectedExecutionException, ScheduledExecutorService, TimeUnit}
-
-import scala.collection.mutable
-import scala.jdk.CollectionConverters._
-import scala.util.control.NonFatal
-
-import org.apache.spark.SparkConf
-import org.apache.spark.internal.Logging
-import org.apache.spark.internal.LogKeys
-import org.apache.spark.internal.config._
-import org.apache.spark.security._
-import org.apache.spark.ui.UIUtils
-import org.apache.spark.util.{ThreadUtils, Utils}
-
-/**
- * Manager for OIDC-based credential propagation on the driver.
- *
- * This class is a sibling of [[HadoopDelegationTokenManager]] and handles the OIDC credential
- * propagation path independently. Both managers run on independent threads and can both be
- * active simultaneously (e.g., a cluster accessing HDFS via Kerberos and S3 via OIDC).
- *
- * Responsibilities:
- * 1. Reads the current identity token via a [[TokenIngestor]]
- * 2. Calls [[CredentialProvider#resolve]] for each configured scheme
- * 3. Serializes the resulting [[UserCredentials]] and invokes the propagation callback
- * 4. Schedules renewal based on `min(identity token expiry, service credential expiry) -
- * safetyMargin`
- * 5. Retries with exponential backoff on failure
- *
- * Intended to be started from `CoarseGrainedSchedulerBackend.start()` when
- * `spark.security.oidc.enabled=true`, independently of
- * `UserGroupInformation.isSecurityEnabled()`.
- *
- * Lifecycle: call `start()` exactly once, then `stop()` to shut down.
- * Calling `start()` after `stop()` is not supported.
- */
-private[spark] class UserCredentialManager(
- sparkConf: SparkConf,
- tokenIngestor: TokenIngestor,
- onCredentialsUpdate: Array[Byte] => Unit) extends Logging {
-
- private val safetyMargin = sparkConf.get(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN)
- private val minInterval = sparkConf.get(SECURITY_OIDC_RENEWAL_MIN_INTERVAL)
-
- // Counter for exponential backoff calculation.
- // Only accessed from the single-thread renewal executor.
- private var consecutiveFailures: Int = 0
- private val maxBackoffMs: Long = UserCredentialManager.MAX_BACKOFF_MS
-
- private var renewalExecutor: ScheduledExecutorService = _
-
- // Snapshot of credential-related configuration at construction time.
- private val credentialConfMap: java.util.Map[String, String] = sparkConf.getAll
- .filter(_._1.startsWith("spark.security.oidc."))
- .toMap.asJava
-
- /**
- * Start the credential manager. Acquires initial credentials and schedules renewal.
- *
- * The initial credential acquisition is fail-fast: if the token cannot be loaded or
- * no credentials can be resolved, an exception is thrown. Subsequent renewal failures
- * are handled with exponential backoff.
- *
- * @return The serialized initial [[UserCredentials]].
- * @throws IllegalStateException if the initial credential acquisition fails.
- */
- def start(): Array[Byte] = {
- require(renewalExecutor == null, "start() must not be called more than once")
-
- // Initial acquisition is fail-fast (no retry/backoff).
- // This ensures the application fails to start if credentials cannot be obtained,
- // rather than running with null credentials.
- val userContext = tokenIngestor.load()
- if (!userContext.isPresent) {
- throw new IllegalStateException(
- "Failed to start UserCredentialManager: identity token file is missing or malformed. " +
- s"Check ${SECURITY_OIDC_IDENTITY_TOKEN_FILE.key} configuration.")
- }
-
- val ctx = userContext.get()
- logInfo(log"Initial identity token loaded for principal " +
- log"${MDC(LogKeys.PRINCIPAL, ctx.getPrincipal)} " +
- log"(issuer: ${MDC(LogKeys.URI, ctx.getIssuer)})")
-
- val (credentials, earliestExpiry) = resolveCredentials(ctx)
- val serialized = UserCredentialManager.serializeUserCredentials(credentials)
-
- // Propagate initial credentials
- onCredentialsUpdate(serialized)
-
- // Create the renewal executor only after successful initial acquisition.
- // This avoids leaking a daemon thread if the fail-fast path throws, and
- // keeps the require() guard valid for retry scenarios.
- renewalExecutor =
- ThreadUtils.newDaemonSingleThreadScheduledExecutor("user-credential-renewal")
-
- // Schedule first renewal
- val renewalDelay = computeRenewalDelay(ctx, earliestExpiry)
- scheduleRenewal(renewalDelay)
-
- logInfo(log"Credential acquisition successful. Next renewal in " +
- log"${MDC(LogKeys.TIME_UNITS, UIUtils.formatDuration(renewalDelay))}.")
- serialized
- }
-
- def stop(): Unit = {
- if (renewalExecutor != null) {
- renewalExecutor.shutdownNow()
- }
- // Close all initialized credential providers to release resources (e.g., HTTP clients).
- // CredentialProviderLoader.closeAll() operates on global static state, which is safe
- // because Spark enforces a single SparkContext (and thus a single UserCredentialManager)
- // per JVM.
- try {
- CredentialProviderLoader.closeAll()
- } catch {
- case e: InterruptedException =>
- Thread.currentThread().interrupt()
- logWarning(log"Interrupted while closing credential providers during shutdown.", e)
- case NonFatal(e) =>
- logWarning(log"Error closing credential providers during shutdown.", e)
- }
- }
-
- /**
- * Scheduled renewal task: load identity token, resolve credentials, propagate,
- * and schedule the next renewal. Failures are retried with exponential backoff.
- */
- private def renewCredentialsTask(): Unit = {
- try {
- val userContext = tokenIngestor.load()
- if (!userContext.isPresent) {
- throw new IllegalStateException(
- "TokenIngestor returned empty - identity token file may be missing or malformed")
- }
-
- val ctx = userContext.get()
- logInfo(log"Loaded identity token for principal " +
- log"${MDC(LogKeys.PRINCIPAL, ctx.getPrincipal)} " +
- log"(issuer: ${MDC(LogKeys.URI, ctx.getIssuer)})")
-
- val (credentials, earliestExpiry) = resolveCredentials(ctx)
- val serialized = UserCredentialManager.serializeUserCredentials(credentials)
-
- // Propagate credentials to executors. Errors here are logged separately
- // so that credential-fetch success is not conflated with distribution failure.
- try {
- onCredentialsUpdate(serialized)
- } catch {
- case e: Exception =>
- logWarning(log"Credentials were resolved successfully but failed to propagate " +
- log"to executors. Executors will receive updated credentials on next renewal.", e)
- }
-
- // Reset backoff on successful credential resolution. This is intentionally done
- // even if onCredentialsUpdate failed above: the backoff counter tracks credential
- // *resolution* failures (STS/provider issues), not propagation failures.
- // Propagation failures are transient and will be retried on next scheduled renewal.
- consecutiveFailures = 0
-
- // Schedule next renewal
- val renewalDelay = computeRenewalDelay(ctx, earliestExpiry)
- scheduleRenewal(renewalDelay)
-
- logInfo(log"Credential renewal successful. Next renewal in " +
- log"${MDC(LogKeys.TIME_UNITS, UIUtils.formatDuration(renewalDelay))}.")
- } catch {
- case _: InterruptedException =>
- // Shutting down, ignore
- case NonFatal(e) =>
- consecutiveFailures += 1
- val failures = consecutiveFailures
- val delay = computeBackoffDelay()
- logWarning(log"Failed to renew user credentials (attempt " +
- log"${MDC(LogKeys.NUM_RETRY, failures)}), " +
- log"will retry in ${MDC(LogKeys.TIME_UNITS, UIUtils.formatDuration(delay))}.", e)
- scheduleRenewal(delay)
- case t: Throwable =>
- // A fatal error would otherwise be swallowed by the executor, silently stopping
- // all future renewals. Log it before propagating.
- logError(log"Fatal error during credential renewal; no further renewals will be " +
- log"scheduled.", t)
- throw t
- }
- }
-
- /**
- * Resolve credentials for all schemes that have registered providers.
- *
- * @return Tuple of (UserCredentials, earliest expiry across all service credentials)
- */
- private def resolveCredentials(
- ctx: UserContext): (UserCredentials, Option[Instant]) = {
- val schemes = discoverSchemes()
-
- val credentialMap = new mutable.HashMap[String, ServiceCredential]()
- var earliestExpiry: Option[Instant] = None
-
- for (scheme <- schemes) {
- try {
- val providerOpt = CredentialProviderLoader.providerFor(scheme, credentialConfMap)
- if (providerOpt.isPresent) {
- val provider = providerOpt.get()
- // Use a synthetic target URI with just the scheme for initial resolution.
- // The full URI is passed in future versions when path-specific resolution is needed.
- // The "synthetic" authority signals this is not a real endpoint but a placeholder
- // for scheme-based provider selection.
- val target = new URI(scheme, UserCredentialManager.SYNTHETIC_TARGET_AUTHORITY,
- "/", null, null)
- val credential = provider.resolve(ctx, target)
- if (credential == null) {
- logWarning(log"Provider for scheme ${MDC(LogKeys.URI, scheme)} " +
- log"returned null; skipping.")
- } else {
- credentialMap.put(scheme, credential)
-
- val expiry = credential.getExpiresAt
- if (expiry != null) {
- earliestExpiry = earliestExpiry match {
- case Some(existing) if existing.isBefore(expiry) => Some(existing)
- case _ => Some(expiry)
- }
- }
- }
- } else {
- logWarning(log"No credential provider found for scheme " +
- log"${MDC(LogKeys.URI, scheme)}. Skipping.")
- }
- } catch {
- case e: Exception =>
- logWarning(log"Failed to resolve credentials for scheme " +
- log"${MDC(LogKeys.URI, scheme)}. Skipping this provider.", e)
- }
- }
-
- if (credentialMap.isEmpty) {
- throw new IllegalStateException(
- "No credential providers resolved any credentials. " +
- "Check that providers are on the classpath and configured correctly.")
- }
-
- (new UserCredentials(credentialMap.asJava), earliestExpiry)
- }
-
- /**
- * Discover all schemes that have at least one registered provider.
- *
- * Schemes are determined by explicit configuration keys of the form
- * `spark.security.oidc.provider.`. If no explicit configuration
- * exists, the method queries `CredentialProviderLoader` to discover all providers
- * registered via ServiceLoader and collects their supported schemes.
- *
- * Note: When using auto-discovery (no explicit config), multiple providers may
- * support the same scheme. In that case, `CredentialProviderLoader.providerFor`
- * will throw an `IllegalArgumentException` for that scheme. The caller
- * (`resolveCredentials`) handles this gracefully via per-provider exception catching,
- * logging a warning and continuing with remaining schemes.
- */
- private def discoverSchemes(): Set[String] = {
- // Extract explicitly configured scheme names.
- // Keys are of the form spark.security.oidc.provider. or
- // spark.security.oidc.provider... We extract only the first
- // segment after the prefix to get the scheme name.
- val providerPrefix = "spark.security.oidc.provider."
- val explicitSchemes = credentialConfMap.asScala
- .filter { case (k, _) => k.startsWith(providerPrefix) }
- .map { case (k, _) => k.stripPrefix(providerPrefix).split('.').head }
- .toSet
-
- if (explicitSchemes.nonEmpty) {
- explicitSchemes
- } else {
- // No explicit scheme configuration. Discover all schemes that have a provider
- // available on the classpath by probing CredentialProviderLoader.
- // This covers both built-in providers (e.g., connector/credential-aws for "s3a")
- // and third-party providers registered via ServiceLoader.
- CredentialProviderLoader.discoverAllSchemes().asScala.toSet
- }
- }
-
- /**
- * Compute the delay until next renewal.
- * Uses min(identity token expiry, service credential expiry) - safetyMargin,
- * bounded below by minInterval.
- */
- private[security] def computeRenewalDelay(
- ctx: UserContext,
- earliestCredentialExpiry: Option[Instant]): Long = {
- val now = System.currentTimeMillis()
-
- // Consider identity token expiry
- val tokenExpiry: Option[Long] = Option(ctx.getExpiresAt).map(_.toEpochMilli)
-
- // Consider earliest service credential expiry
- val credExpiry: Option[Long] = earliestCredentialExpiry.map(_.toEpochMilli)
-
- // Take the minimum of both
- val effectiveExpiry: Option[Long] = (tokenExpiry, credExpiry) match {
- case (Some(t), Some(c)) => Some(math.min(t, c))
- case (Some(t), None) => Some(t)
- case (None, Some(c)) => Some(c)
- case (None, None) => None
- }
-
- effectiveExpiry match {
- case Some(expiry) =>
- math.max(expiry - now - safetyMargin, minInterval)
- case None =>
- // No expiry information available from either the identity token or service
- // credentials. Use half of the default suggested TTL (15 min / 2 = 7.5 min,
- // rounded down) as a conservative polling interval. This ensures credentials
- // are refreshed even when providers don't report expiration times.
- UserCredentialManager.DEFAULT_RENEWAL_INTERVAL_NO_EXPIRY_MS
- }
- }
-
- /**
- * Compute backoff delay using exponential backoff with jitter.
- * Bounded by minInterval (floor) and maxBackoffMs (ceiling).
- */
- private[security] def computeBackoffDelay(): Long = {
- // Guard against negative or zero shift amounts. consecutiveFailures should always
- // be >= 1 when this is called (incremented before calling), but we protect against
- // edge cases defensively.
- val shiftAmount = math.max(0, math.min(consecutiveFailures - 1, 6))
- val baseDelay = minInterval * (1L << shiftAmount)
- val cappedDelay = math.min(baseDelay, maxBackoffMs)
- // Add 10% jitter to avoid thundering herd on recovery
- val jitter = (cappedDelay * 0.1 * math.random()).toLong
- math.max(cappedDelay + jitter, minInterval)
- }
-
- private def scheduleRenewal(delay: Long): Unit = {
- try {
- val renewalTask = new Runnable {
- override def run(): Unit = {
- renewCredentialsTask()
- }
- }
- renewalExecutor.schedule(renewalTask, delay, TimeUnit.MILLISECONDS)
- } catch {
- case _: RejectedExecutionException =>
- // Executor has been shut down (e.g., stop() called concurrently). This is expected
- // during application shutdown -- no further renewals will be scheduled.
- logDebug(log"Renewal scheduling rejected - executor is shut down.")
- }
- }
-
-}
-
-private[spark] object UserCredentialManager {
-
- /**
- * Synthetic authority used in target URIs for scheme-based provider resolution.
- * Providers should not rely on this value; it signals that no specific endpoint
- * is targeted and only the URI scheme is meaningful.
- */
- private val SYNTHETIC_TARGET_AUTHORITY = "synthetic"
-
- /**
- * Maximum backoff delay between credential renewal retry attempts.
- * Caps the exponential backoff to prevent excessively long gaps between retries.
- */
- private val MAX_BACKOFF_MS: Long = TimeUnit.MINUTES.toMillis(10)
-
- /**
- * Default renewal interval when neither the identity token nor service credentials
- * report an expiration time. Set to 7 minutes (half of the default 15-minute
- * suggested TTL from CredentialProvider.suggestedTtl()).
- */
- private val DEFAULT_RENEWAL_INTERVAL_NO_EXPIRY_MS: Long = TimeUnit.MINUTES.toMillis(7)
-
- /**
- * ObjectInputFilter pattern restricting deserialization to only the classes needed
- * for UserCredentials. This prevents deserialization gadget chain attacks while
- * allowing Java's built-in collection internal classes and arrays that HashMap uses.
- */
- private val DESERIALIZATION_FILTER: String =
- "org.apache.spark.security.**;" +
- "java.util.**;" +
- "java.time.**;" +
- "java.lang.**;" +
- "maxdepth=10;" +
- "maxarray=1000;" +
- "maxrefs=1000;" +
- "!*"
-
- /**
- * Create a UserCredentialManager if OIDC credential propagation is enabled.
- *
- * @param sparkConf The Spark configuration
- * @param onCredentialsUpdate Callback to propagate credentials to executors
- * @return Some(manager) if enabled, None otherwise
- */
- def create(
- sparkConf: SparkConf,
- onCredentialsUpdate: Array[Byte] => Unit): Option[UserCredentialManager] = {
- if (!sparkConf.get(SECURITY_OIDC_ENABLED)) {
- None
- } else {
- val tokenFile = sparkConf.get(SECURITY_OIDC_IDENTITY_TOKEN_FILE).getOrElse {
- throw new IllegalArgumentException(
- s"${SECURITY_OIDC_IDENTITY_TOKEN_FILE.key} must be set when " +
- s"${SECURITY_OIDC_ENABLED.key} is true")
- }
-
- val tokenIngestor = new FileTokenIngestor(Paths.get(tokenFile))
- Some(new UserCredentialManager(sparkConf, tokenIngestor, onCredentialsUpdate))
- }
- }
-
- /**
- * Serialize [[UserCredentials]] to a byte array using Java serialization.
- */
- private[security] def serializeUserCredentials(credentials: UserCredentials): Array[Byte] = {
- val bos = new ByteArrayOutputStream()
- Utils.tryWithResource(new ObjectOutputStream(bos)) { oos =>
- oos.writeObject(credentials)
- oos.flush()
- }
- bos.toByteArray
- }
-
- /**
- * Deserialize [[UserCredentials]] from a byte array.
- *
- * Uses an [[ObjectInputFilter]] to restrict deserialized classes to only those
- * required for [[UserCredentials]], preventing deserialization attacks.
- */
- def deserializeUserCredentials(bytes: Array[Byte]): UserCredentials = {
- val bis = new java.io.ByteArrayInputStream(bytes)
- Utils.tryWithResource(new ObjectInputStream(bis)) { ois =>
- ois.setObjectInputFilter(
- ObjectInputFilter.Config.createFilter(DESERIALIZATION_FILTER))
- ois.readObject().asInstanceOf[UserCredentials]
- }
- }
-}
diff --git a/core/src/main/scala/org/apache/spark/internal/config/package.scala b/core/src/main/scala/org/apache/spark/internal/config/package.scala
index f4b3b95efe3ba..f3fcac7353329 100644
--- a/core/src/main/scala/org/apache/spark/internal/config/package.scala
+++ b/core/src/main/scala/org/apache/spark/internal/config/package.scala
@@ -1690,46 +1690,6 @@ package object config {
.timeConf(TimeUnit.SECONDS)
.createWithDefaultString("1h")
- private[spark] val SECURITY_OIDC_ENABLED =
- ConfigBuilder("spark.security.oidc.enabled")
- .doc("Whether to enable OIDC credential propagation. When enabled, the driver reads an " +
- "identity token from a file, exchanges it for short-lived service credentials via " +
- "CredentialProvider implementations, and propagates those credentials to executors.")
- .version("4.3.0")
- .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
- .booleanConf
- .createWithDefault(false)
-
- private[spark] val SECURITY_OIDC_IDENTITY_TOKEN_FILE =
- ConfigBuilder("spark.security.oidc.identityToken.file")
- .doc("Path to the OIDC identity token file on the driver. Required when " +
- "spark.security.oidc.enabled is true. The file should contain a JWT token " +
- "(e.g., a Kubernetes projected service account token).")
- .version("4.3.0")
- .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
- .stringConf
- .createOptional
-
- private[spark] val SECURITY_OIDC_RENEWAL_SAFETY_MARGIN =
- ConfigBuilder("spark.security.oidc.renewal.safetyMargin")
- .doc("How long before credential expiry to trigger renewal. Credentials are refreshed " +
- "at min(identity token expiry, service credential expiry) minus this margin.")
- .version("4.3.0")
- .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
- .timeConf(TimeUnit.MILLISECONDS)
- .checkValue(_ > 0, "The safety margin must be a positive time value.")
- .createWithDefaultString("60s")
-
- private[spark] val SECURITY_OIDC_RENEWAL_MIN_INTERVAL =
- ConfigBuilder("spark.security.oidc.renewal.minInterval")
- .doc("Minimum interval between credential renewal attempts. This prevents tight renewal " +
- "loops when credentials have very short TTLs or when failures cause rapid retries.")
- .version("4.3.0")
- .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
- .timeConf(TimeUnit.MILLISECONDS)
- .checkValue(_ > 0, "The minimum renewal interval must be a positive time value.")
- .createWithDefaultString("30s")
-
private[spark] val DIRECT_CREDENTIAL_PROVIDERS_ENABLED =
ConfigBuilder("spark.security.directCredentialProviders.enabled")
.doc(
diff --git a/core/src/test/java/org/apache/spark/security/AnotherFakeCredentialProvider.java b/core/src/test/java/org/apache/spark/security/AnotherFakeCredentialProvider.java
deleted file mode 100644
index b4f3ae40700b0..0000000000000
--- a/core/src/test/java/org/apache/spark/security/AnotherFakeCredentialProvider.java
+++ /dev/null
@@ -1,61 +0,0 @@
-/*
- * 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.security;
-
-import java.net.URI;
-import java.time.Instant;
-import java.util.Map;
-import java.util.Set;
-
-/**
- * A second fake credential provider for testing. Supports only the "shared" scheme
- * to create ambiguity with {@link FakeCredentialProvider}.
- */
-public class AnotherFakeCredentialProvider implements CredentialProvider {
-
- /** Sentinel URI host that triggers a CredentialResolutionException. */
- public static final String ERROR_HOST = "error.example.com";
-
- private Map initConf;
-
- @Override
- public void init(Map conf) {
- this.initConf = conf;
- }
-
- @Override
- public Set supportedSchemes() {
- return Set.of("shared");
- }
-
- @Override
- public ServiceCredential resolve(UserContext user, URI target)
- throws CredentialResolutionException {
- if (target.getHost() != null && target.getHost().equals(ERROR_HOST)) {
- throw new CredentialResolutionException(
- "Simulated failure from AnotherFakeCredentialProvider for target: " + target);
- }
- Instant expiresAt = Instant.now().plus(suggestedTtl());
- return new ServiceCredential(Map.of("provider", "another"), expiresAt);
- }
-
- /** Returns the configuration map passed to {@link #init(Map)}, or null if not yet called. */
- public Map getInitConf() {
- return initConf;
- }
-}
diff --git a/core/src/test/java/org/apache/spark/security/CredentialProviderLoaderSuite.java b/core/src/test/java/org/apache/spark/security/CredentialProviderLoaderSuite.java
deleted file mode 100644
index 34261f2524c2c..0000000000000
--- a/core/src/test/java/org/apache/spark/security/CredentialProviderLoaderSuite.java
+++ /dev/null
@@ -1,487 +0,0 @@
-/*
- * 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.security;
-
-import java.net.URI;
-import java.time.Duration;
-import java.time.Instant;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Set;
-
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertFalse;
-import static org.junit.jupiter.api.Assertions.assertInstanceOf;
-import static org.junit.jupiter.api.Assertions.assertNotNull;
-import static org.junit.jupiter.api.Assertions.assertSame;
-import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-
-/**
- * Tests for {@link CredentialProviderLoader} covering ServiceLoader discovery,
- * single-candidate resolution, ambiguity handling, explicit selection, and error cases.
- */
-public class CredentialProviderLoaderSuite {
-
- @BeforeEach
- public void setUp() {
- CredentialProviderLoader.resetForTesting();
- }
-
- @Test
- public void testServiceLoaderDiscoversFakeProviders() {
- // The "fake" scheme is supported only by FakeCredentialProvider (single candidate).
- // If discovery works, providerFor should find it.
- Map conf = Map.of();
- Optional result = CredentialProviderLoader.providerFor("fake", conf);
- assertTrue(result.isPresent(), "ServiceLoader should discover FakeCredentialProvider");
- assertInstanceOf(FakeCredentialProvider.class, result.get());
- }
-
- @Test
- public void testSingleCandidateSchemeResolvesWithNoConf() {
- // "fake" is supported only by FakeCredentialProvider
- Map conf = Map.of();
- Optional result = CredentialProviderLoader.providerFor("fake", conf);
- assertTrue(result.isPresent());
- assertInstanceOf(FakeCredentialProvider.class, result.get());
- }
-
- @Test
- public void testSharedSchemeWithNoConfThrowsAmbiguity() {
- // "shared" is supported by both FakeCredentialProvider and AnotherFakeCredentialProvider
- Map conf = Map.of();
- IllegalArgumentException e = assertThrows(IllegalArgumentException.class,
- () -> CredentialProviderLoader.providerFor("shared", conf));
- assertTrue(e.getMessage().contains("Multiple credential providers"),
- "Should mention multiple providers: " + e.getMessage());
- assertTrue(e.getMessage().contains("shared"),
- "Should mention the scheme: " + e.getMessage());
- assertTrue(e.getMessage().contains("spark.security.oidc.provider.shared"),
- "Should mention the config key: " + e.getMessage());
- assertTrue(e.getMessage().contains(FakeCredentialProvider.class.getName()),
- "Should list FakeCredentialProvider: " + e.getMessage());
- assertTrue(e.getMessage().contains(AnotherFakeCredentialProvider.class.getName()),
- "Should list AnotherFakeCredentialProvider: " + e.getMessage());
- }
-
- @Test
- public void testEmptyStringConfTreatedAsUnsetThrowsAmbiguity() {
- // An empty-string value for the explicit provider conf key should be equivalent to unset,
- // meaning the ambiguity error is still raised for multi-candidate schemes.
- Map conf = new HashMap<>();
- conf.put("spark.security.oidc.provider.shared", "");
- IllegalArgumentException e = assertThrows(IllegalArgumentException.class,
- () -> CredentialProviderLoader.providerFor("shared", conf));
- assertTrue(e.getMessage().contains("Multiple credential providers"),
- "Empty conf value should behave as unset: " + e.getMessage());
- }
-
- @Test
- public void testSharedSchemeWithExplicitConfSelectsFake() {
- Map conf = Map.of(
- "spark.security.oidc.provider.shared",
- FakeCredentialProvider.class.getName());
- Optional result = CredentialProviderLoader.providerFor("shared", conf);
- assertTrue(result.isPresent());
- assertInstanceOf(FakeCredentialProvider.class, result.get());
- }
-
- @Test
- public void testSharedSchemeWithExplicitConfSelectsAnother() {
- Map conf = Map.of(
- "spark.security.oidc.provider.shared",
- AnotherFakeCredentialProvider.class.getName());
- Optional result = CredentialProviderLoader.providerFor("shared", conf);
- assertTrue(result.isPresent());
- assertInstanceOf(AnotherFakeCredentialProvider.class, result.get());
- }
-
- @Test
- public void testConfNamingUnknownClassThrowsClearError() {
- Map conf = Map.of(
- "spark.security.oidc.provider.fake",
- "com.example.NonExistentProvider");
- IllegalArgumentException e = assertThrows(IllegalArgumentException.class,
- () -> CredentialProviderLoader.providerFor("fake", conf));
- assertTrue(e.getMessage().contains("com.example.NonExistentProvider"),
- "Should mention the configured class: " + e.getMessage());
- assertTrue(e.getMessage().contains("fake"),
- "Should mention the scheme: " + e.getMessage());
- assertTrue(e.getMessage().contains(FakeCredentialProvider.class.getName()),
- "Should list available candidate(s): " + e.getMessage());
- }
-
- @Test
- public void testConfNamingNonSupportingClassThrowsClearError() {
- // AnotherFakeCredentialProvider does NOT support "fake" scheme
- Map conf = Map.of(
- "spark.security.oidc.provider.fake",
- AnotherFakeCredentialProvider.class.getName());
- IllegalArgumentException e = assertThrows(IllegalArgumentException.class,
- () -> CredentialProviderLoader.providerFor("fake", conf));
- assertTrue(e.getMessage().contains(AnotherFakeCredentialProvider.class.getName()),
- "Should mention the configured class: " + e.getMessage());
- assertTrue(e.getMessage().contains("fake"),
- "Should mention the scheme: " + e.getMessage());
- assertTrue(e.getMessage().contains(FakeCredentialProvider.class.getName()),
- "Should list available candidate(s): " + e.getMessage());
- }
-
- @Test
- public void testSingleCandidateWithCorrectExplicitConfSelectsIt() {
- // "fake" is supported only by FakeCredentialProvider; conf names the correct class.
- Map conf = Map.of(
- "spark.security.oidc.provider.fake",
- FakeCredentialProvider.class.getName());
- Optional result = CredentialProviderLoader.providerFor("fake", conf);
- assertTrue(result.isPresent());
- assertInstanceOf(FakeCredentialProvider.class, result.get());
- }
-
- @Test
- public void testSingleCandidateWithWrongExplicitConfThrowsClearError() {
- // "fake" is supported only by FakeCredentialProvider but conf names a different class.
- // This validates that explicit conf is enforced even for single-candidate schemes.
- Map conf = Map.of(
- "spark.security.oidc.provider.fake",
- "org.apache.spark.security.SomeOtherProvider");
- IllegalArgumentException e = assertThrows(IllegalArgumentException.class,
- () -> CredentialProviderLoader.providerFor("fake", conf));
- assertTrue(e.getMessage().contains("fake"),
- "Should mention the scheme: " + e.getMessage());
- assertTrue(e.getMessage().contains("org.apache.spark.security.SomeOtherProvider"),
- "Should mention the configured class: " + e.getMessage());
- assertTrue(e.getMessage().contains(FakeCredentialProvider.class.getName()),
- "Should list the available candidate: " + e.getMessage());
- }
-
- @Test
- public void testUnknownSchemeReturnsEmpty() {
- Map conf = Map.of();
- Optional result =
- CredentialProviderLoader.providerFor("nonexistent", conf);
- assertFalse(result.isPresent(), "Unknown scheme should return empty");
- }
-
- @Test
- public void testInitConfIsInvokedOnSelectedProvider() {
- Map conf = new HashMap<>();
- conf.put("spark.security.oidc.endpoint", "https://sts.example.com");
- conf.put("spark.security.oidc.roleArn", "arn:aws:iam::123456:role/test");
-
- Optional result = CredentialProviderLoader.providerFor("fake", conf);
- assertTrue(result.isPresent());
- FakeCredentialProvider fake = (FakeCredentialProvider) result.get();
- assertNotNull(fake.getInitConf(), "init() should have been called");
- assertEquals("https://sts.example.com",
- fake.getInitConf().get("spark.security.oidc.endpoint"));
- assertEquals("arn:aws:iam::123456:role/test",
- fake.getInitConf().get("spark.security.oidc.roleArn"));
- }
-
- @Test
- public void testProviderInitializedExactlyOnce() {
- // Call providerFor twice for the same scheme and assert:
- // (a) the SAME provider instance is returned
- // (b) init was invoked EXACTLY ONCE
- Map conf1 = new HashMap<>();
- conf1.put("spark.security.oidc.tag", "first-call");
- Map conf2 = new HashMap<>();
- conf2.put("spark.security.oidc.tag", "second-call");
-
- Optional result1 = CredentialProviderLoader.providerFor("fake", conf1);
- Optional result2 = CredentialProviderLoader.providerFor("fake", conf2);
-
- assertTrue(result1.isPresent());
- assertTrue(result2.isPresent());
- assertSame(result1.get(), result2.get(),
- "providerFor should return the same cached instance");
-
- FakeCredentialProvider fake = (FakeCredentialProvider) result1.get();
- assertEquals(1, fake.getInitCount(),
- "init() should be called exactly once (first-conf-wins)");
- assertEquals("first-call", fake.getInitConf().get("spark.security.oidc.tag"),
- "First call's conf should win");
- }
-
- @Test
- public void testNullSupportedSchemesThrowsClearError() {
- // Inject a provider that returns null from supportedSchemes() to verify the guard.
- CredentialProvider nullSchemesProvider = new CredentialProvider() {
- @Override
- public void init(Map conf) {}
-
- @Override
- public Set supportedSchemes() {
- return null;
- }
-
- @Override
- public ServiceCredential resolve(UserContext user, URI target) {
- return null;
- }
- };
- CredentialProviderLoader.setProvidersForTesting(
- List.of(nullSchemesProvider));
-
- IllegalStateException e = assertThrows(IllegalStateException.class,
- () -> CredentialProviderLoader.providerFor("anything", Map.of()));
- assertTrue(e.getMessage().contains("returned null from supportedSchemes()"),
- "Should have a clear null-schemes message: " + e.getMessage());
- }
-
- @Test
- public void testResolveReturnsExpectedServiceCredential() throws Exception {
- Map conf = Map.of();
- Optional result = CredentialProviderLoader.providerFor("fake", conf);
- assertTrue(result.isPresent());
-
- UserContext user = new UserContext(
- "testuser", "https://idp.example.com", "token", Instant.now(), null);
- URI target = URI.create("fake://bucket/path");
- ServiceCredential cred = result.get().resolve(user, target);
-
- assertNotNull(cred);
- assertEquals("fake", cred.getProperties().get("provider"));
- assertNotNull(cred.getExpiresAt(), "expiresAt should be set");
- }
-
- @Test
- public void testResolveSentinelThrowsCredentialResolutionException() {
- Map conf = Map.of();
- Optional result = CredentialProviderLoader.providerFor("fake", conf);
- assertTrue(result.isPresent());
-
- UserContext user = new UserContext(
- "testuser", "https://idp.example.com", "token", Instant.now(), null);
- URI errorTarget = URI.create("fake://error.example.com/path");
-
- CredentialResolutionException e = assertThrows(CredentialResolutionException.class,
- () -> result.get().resolve(user, errorTarget));
- assertTrue(e.getMessage().contains("error.example.com"),
- "Exception should reference the target: " + e.getMessage());
- }
-
- @Test
- public void testSchemeNormalizationIsCaseInsensitive() {
- // "FAKE" should resolve the same as "fake"
- Map conf = Map.of();
- Optional result = CredentialProviderLoader.providerFor("FAKE", conf);
- assertTrue(result.isPresent());
- assertInstanceOf(FakeCredentialProvider.class, result.get());
- }
-
- @Test
- public void testExplicitSelectionWithUppercaseSchemeNormalizesConfKey() {
- // The conf key uses normalized (lowercase) scheme
- Map conf = Map.of(
- "spark.security.oidc.provider.shared",
- FakeCredentialProvider.class.getName());
- Optional result = CredentialProviderLoader.providerFor("SHARED", conf);
- assertTrue(result.isPresent());
- assertInstanceOf(FakeCredentialProvider.class, result.get());
- }
-
- @Test
- public void testNullSchemeThrowsNPE() {
- NullPointerException e = assertThrows(NullPointerException.class,
- () -> CredentialProviderLoader.providerFor(null, Map.of()));
- assertTrue(e.getMessage().contains("scheme must not be null"),
- "Should have a clear message: " + e.getMessage());
- }
-
- @Test
- public void testNullConfThrowsNPE() {
- NullPointerException e = assertThrows(NullPointerException.class,
- () -> CredentialProviderLoader.providerFor("fake", null));
- assertTrue(e.getMessage().contains("conf must not be null"),
- "Should have a clear message: " + e.getMessage());
- }
-
- @Test
- public void testSuggestedTtlDefaultValue() {
- Map conf = Map.of();
- Optional result = CredentialProviderLoader.providerFor("fake", conf);
- assertTrue(result.isPresent());
- assertEquals(Duration.ofMinutes(15), result.get().suggestedTtl());
- }
-
- @Test
- public void testInitConfScopedToOidcKeysOnly() {
- // Verify that init() receives only spark.security.oidc.* keys,
- // and foreign secrets from other subsystems are NOT leaked to providers.
- Map conf = new HashMap<>();
- conf.put("spark.security.oidc.provider.fake",
- FakeCredentialProvider.class.getName());
- conf.put("spark.security.oidc.endpoint", "https://sts.example.com");
- conf.put("spark.authenticate.secret", "TOPSECRET");
- conf.put("spark.ssl.keyPassword", "keypass");
-
- Optional result = CredentialProviderLoader.providerFor("fake", conf);
- assertTrue(result.isPresent());
- FakeCredentialProvider fake = (FakeCredentialProvider) result.get();
- Map initConf = fake.getInitConf();
- assertNotNull(initConf, "init() should have been called");
-
- // OIDC keys should be present
- assertEquals("https://sts.example.com",
- initConf.get("spark.security.oidc.endpoint"));
- assertTrue(initConf.containsKey("spark.security.oidc.provider.fake"),
- "Provider selection key should be included (it starts with the prefix)");
-
- // Foreign secrets must NOT be present
- assertFalse(initConf.containsKey("spark.authenticate.secret"),
- "Foreign secret should not leak to provider init()");
- assertFalse(initConf.containsKey("spark.ssl.keyPassword"),
- "Foreign secret should not leak to provider init()");
- assertFalse(initConf.values().contains("TOPSECRET"),
- "Foreign secret value should not leak to provider init()");
- }
-
- @Test
- public void testEmptySchemeThrowsIllegalArgument() {
- IllegalArgumentException e = assertThrows(IllegalArgumentException.class,
- () -> CredentialProviderLoader.providerFor("", Map.of()));
- assertEquals("scheme must not be empty", e.getMessage());
- }
-
- @Test
- public void testCloseAllClosesInitializedProviders() throws Exception {
- // Initialize a provider by calling providerFor
- Map conf = Map.of();
- Optional result = CredentialProviderLoader.providerFor("fake", conf);
- assertTrue(result.isPresent());
- FakeCredentialProvider fake = (FakeCredentialProvider) result.get();
- assertEquals(0, fake.getCloseCount(), "close() not yet called");
-
- // Call closeAll
- CredentialProviderLoader.closeAll();
-
- assertEquals(1, fake.getCloseCount(), "close() should be called exactly once");
- }
-
- @Test
- public void testCloseAllSuppressesExceptionsAndClosesAll() throws Exception {
- // Two providers: first throws on close, second should still be closed
- CredentialProvider throwingProvider = new CredentialProvider() {
- @Override
- public void init(Map conf) {}
-
- @Override
- public Set supportedSchemes() {
- return Set.of("throwing");
- }
-
- @Override
- public ServiceCredential resolve(UserContext user, URI target) {
- return new ServiceCredential(Map.of(), Instant.now().plusSeconds(60));
- }
-
- @Override
- public void close() throws Exception {
- throw new RuntimeException("Simulated close failure");
- }
- };
-
- FakeCredentialProvider fakeProvider = new FakeCredentialProvider();
-
- CredentialProviderLoader.setProvidersForTesting(
- List.of(throwingProvider, fakeProvider));
-
- // Initialize both by selecting them
- Map conf = new HashMap<>();
- conf.put("spark.security.oidc.provider.throwing", throwingProvider.getClass().getName());
- CredentialProviderLoader.providerFor("throwing", conf);
- CredentialProviderLoader.providerFor("fake", conf);
-
- // closeAll should throw (from throwingProvider) but still close fakeProvider
- Exception e = assertThrows(Exception.class,
- () -> CredentialProviderLoader.closeAll());
- assertTrue(e.getMessage().contains("Simulated close failure"));
- assertEquals(1, fakeProvider.getCloseCount(),
- "Second provider should still be closed even when first throws");
- }
-
- @Test
- public void testCloseAllWithNoInitializedProvidersIsNoOp() throws Exception {
- // No providers initialized — closeAll should not throw
- CredentialProviderLoader.closeAll();
- // If we reach here, no exception was thrown — success
- }
-
- @Test
- public void testInitRetryAfterFailure() {
- // A provider whose init() throws on the first call then succeeds on the second.
- // First providerFor should propagate the failure; second providerFor should retry
- // init() and succeed.
- CredentialProvider failOnceThenSucceed = new CredentialProvider() {
- private int initAttempts = 0;
- private boolean initialized = false;
-
- @Override
- public void init(Map conf) {
- initAttempts++;
- if (initAttempts == 1) {
- throw new RuntimeException("Simulated transient init failure");
- }
- initialized = true;
- }
-
- @Override
- public Set supportedSchemes() {
- return Set.of("retryscheme");
- }
-
- @Override
- public ServiceCredential resolve(UserContext user, URI target) {
- return new ServiceCredential(Map.of("ok", "true"), Instant.now().plusSeconds(60));
- }
-
- @Override
- public String toString() {
- return "initAttempts=" + initAttempts + ",initialized=" + initialized;
- }
- };
-
- CredentialProviderLoader.setProvidersForTesting(List.of(failOnceThenSucceed));
-
- // First call: init() throws, providerFor should propagate
- Map conf = Map.of();
- RuntimeException e = assertThrows(RuntimeException.class,
- () -> CredentialProviderLoader.providerFor("retryscheme", conf));
- assertTrue(e.getMessage().contains("Simulated transient init failure"));
-
- // Second call: init() should be retried and succeed
- Optional result =
- CredentialProviderLoader.providerFor("retryscheme", conf);
- assertTrue(result.isPresent(), "Second providerFor should succeed after init retry");
-
- // Verify init was called exactly twice (proving the retry)
- String state = result.get().toString();
- assertTrue(state.contains("initAttempts=2"),
- "init() should have been invoked twice: " + state);
- assertTrue(state.contains("initialized=true"),
- "Provider should be initialized after retry: " + state);
- }
-}
diff --git a/core/src/test/java/org/apache/spark/security/CredentialTypesSuite.java b/core/src/test/java/org/apache/spark/security/CredentialTypesSuite.java
deleted file mode 100644
index 59d8ed66135da..0000000000000
--- a/core/src/test/java/org/apache/spark/security/CredentialTypesSuite.java
+++ /dev/null
@@ -1,317 +0,0 @@
-/*
- * 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.security;
-
-import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
-import java.io.NotSerializableException;
-import java.io.ObjectInputStream;
-import java.io.ObjectOutputStream;
-import java.time.Instant;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Optional;
-
-import org.junit.jupiter.api.Test;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertFalse;
-import static org.junit.jupiter.api.Assertions.assertNotEquals;
-import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-
-public class CredentialTypesSuite {
-
- private static final String TOKEN = "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.secret";
- private static final Instant NOW = Instant.parse("2026-01-15T12:00:00Z");
- private static final Instant PAST = Instant.parse("2026-01-15T11:00:00Z");
- private static final Instant FUTURE = Instant.parse("2026-01-15T13:00:00Z");
-
- // --- UserContext tests ---
-
- @Test
- public void testUserContextGettersRoundTrip() {
- UserContext ctx = new UserContext("user1", "https://idp.example.com", TOKEN, NOW, FUTURE);
- assertEquals("user1", ctx.getPrincipal());
- assertEquals("https://idp.example.com", ctx.getIssuer());
- assertEquals(TOKEN, ctx.getRawToken());
- assertEquals(NOW, ctx.getIssuedAt());
- assertEquals(FUTURE, ctx.getExpiresAt());
- }
-
- @Test
- public void testUserContextToStringRedactsToken() {
- UserContext ctx = new UserContext("user1", "https://idp.example.com", TOKEN, NOW, FUTURE);
- String str = ctx.toString();
- assertFalse(str.contains(TOKEN), "toString must not contain the raw token value");
- assertTrue(str.contains("[REDACTED]"), "toString must contain [REDACTED]");
- assertTrue(str.contains("user1"));
- assertTrue(str.contains("https://idp.example.com"));
- }
-
- @Test
- public void testUserContextIsExpiredBoundary() {
- // expiresAt == NOW -> expired (now >= expiresAt)
- UserContext ctx = new UserContext("u", "iss", TOKEN, PAST, NOW);
- assertTrue(ctx.isExpired(NOW), "Should be expired when now == expiresAt");
- assertFalse(ctx.isExpired(PAST), "Should not be expired when now < expiresAt");
- assertTrue(ctx.isExpired(FUTURE), "Should be expired when now > expiresAt");
- }
-
- @Test
- public void testUserContextIsExpiredWithNullExpiry() {
- UserContext ctx = new UserContext("u", "iss", TOKEN, NOW, null);
- assertFalse(ctx.isExpired(NOW), "Should not be expired when expiresAt is null");
- }
-
- @Test
- public void testUserContextEqualsAndHashCode() {
- UserContext a = new UserContext("u", "iss", TOKEN, NOW, FUTURE);
- UserContext b = new UserContext("u", "iss", TOKEN, NOW, FUTURE);
- UserContext c = new UserContext("other", "iss", TOKEN, NOW, FUTURE);
- assertEquals(a, b);
- assertEquals(a.hashCode(), b.hashCode());
- assertNotEquals(a, c);
- }
-
- @Test
- public void testUserContextEqualsExcludesRawToken() {
- // rawToken is secret material and intentionally excluded from equals/hashCode:
- // two contexts that differ only by rawToken are considered equal.
- UserContext a = new UserContext("u", "iss", "token-A", NOW, FUTURE);
- UserContext b = new UserContext("u", "iss", "token-B", NOW, FUTURE);
- assertEquals(a, b, "equals must not depend on rawToken");
- assertEquals(a.hashCode(), b.hashCode(), "hashCode must not depend on rawToken");
- // A differing identity field still breaks equality.
- UserContext c = new UserContext("u", "other-iss", "token-A", NOW, FUTURE);
- assertNotEquals(a, c);
- }
-
- @Test
- public void testUserContextNotSerializable() {
- UserContext ctx = new UserContext("user1", "https://idp.example.com", TOKEN, NOW, FUTURE);
- ByteArrayOutputStream bos = new ByteArrayOutputStream();
- assertThrows(NotSerializableException.class, () -> {
- try (ObjectOutputStream oos = new ObjectOutputStream(bos)) {
- oos.writeObject(ctx);
- }
- });
- }
-
- // --- ServiceCredential tests ---
-
- @Test
- public void testServiceCredentialDefensiveCopy() {
- Map props = new HashMap<>();
- props.put("key", "value");
- ServiceCredential cred = new ServiceCredential(props, FUTURE);
- // Mutating the original map must not affect the credential
- props.put("key2", "value2");
- assertFalse(cred.getProperties().containsKey("key2"));
- assertEquals(1, cred.getProperties().size());
- }
-
- @Test
- public void testServiceCredentialIsExpired() {
- ServiceCredential cred = new ServiceCredential(Map.of("k", "v"), NOW);
- assertTrue(cred.isExpired(NOW), "Should be expired when now == expiresAt");
- assertFalse(cred.isExpired(PAST), "Should not be expired when now < expiresAt");
- assertTrue(cred.isExpired(FUTURE), "Should be expired when now > expiresAt");
- }
-
- @Test
- public void testServiceCredentialIsExpiredWithNullExpiry() {
- ServiceCredential cred = new ServiceCredential(Map.of("k", "v"), null);
- assertFalse(cred.isExpired(NOW));
- }
-
- @Test
- public void testServiceCredentialSerializationRoundTrip() throws Exception {
- Map props = new HashMap<>();
- props.put("fs.s3a.access.key", "AKIAIOSFODNN7EXAMPLE");
- props.put("fs.s3a.secret.key", "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLE");
- ServiceCredential original = new ServiceCredential(props, FUTURE);
-
- byte[] bytes = serialize(original);
- ServiceCredential deserialized = (ServiceCredential) deserialize(bytes);
-
- assertEquals(original, deserialized);
- assertEquals(original.getProperties(), deserialized.getProperties());
- assertEquals(original.getExpiresAt(), deserialized.getExpiresAt());
- }
-
- @Test
- public void testServiceCredentialEqualsAndHashCode() {
- ServiceCredential a = new ServiceCredential(Map.of("k", "v"), FUTURE);
- ServiceCredential b = new ServiceCredential(Map.of("k", "v"), FUTURE);
- ServiceCredential c = new ServiceCredential(Map.of("k", "other"), FUTURE);
- assertEquals(a, b);
- assertEquals(a.hashCode(), b.hashCode());
- assertNotEquals(a, c);
- }
-
- @Test
- public void testServiceCredentialToStringRedactsValues() {
- Map props = new HashMap<>();
- props.put("accessKey", "AKIAIOSFODNN7EXAMPLE");
- props.put("secretKey", "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLE");
- ServiceCredential cred = new ServiceCredential(props, FUTURE);
- String str = cred.toString();
- // Secret values must NOT appear
- assertFalse(str.contains("AKIAIOSFODNN7EXAMPLE"),
- "toString must not contain secret property values");
- assertFalse(str.contains("wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLE"),
- "toString must not contain secret property values");
- // Keys SHOULD appear for debuggability
- assertTrue(str.contains("accessKey"), "toString should contain property key names");
- assertTrue(str.contains("secretKey"), "toString should contain property key names");
- // Values are replaced with [REDACTED]
- assertTrue(str.contains("[REDACTED]"), "toString must contain [REDACTED] for values");
- // expiresAt should still appear
- assertTrue(str.contains("expiresAt"), "toString should contain expiresAt");
- }
-
- @Test
- public void testServiceCredentialToStringEmptyProperties() {
- ServiceCredential cred = new ServiceCredential(Map.of(), null);
- String str = cred.toString();
- assertTrue(str.contains("properties={}"),
- "toString should handle empty properties map");
- assertTrue(str.contains("expiresAt=null"));
- }
-
- // --- UserCredentials tests ---
-
- @Test
- public void testUserCredentialsForSchemePresent() {
- ServiceCredential s3Cred = new ServiceCredential(Map.of("k", "v"), FUTURE);
- UserCredentials uc = new UserCredentials(Map.of("s3a", s3Cred));
- Optional result = uc.forScheme("s3a");
- assertTrue(result.isPresent());
- assertEquals(s3Cred, result.get());
- }
-
- @Test
- public void testUserCredentialsForSchemeAbsent() {
- ServiceCredential s3Cred = new ServiceCredential(Map.of("k", "v"), FUTURE);
- UserCredentials uc = new UserCredentials(Map.of("s3a", s3Cred));
- Optional result = uc.forScheme("abfss");
- assertFalse(result.isPresent());
- }
-
- @Test
- public void testUserCredentialsDoesNotExposeToken() {
- // UserCredentials only contains ServiceCredentials, not UserContext or tokens
- ServiceCredential cred = new ServiceCredential(Map.of("token", "short-lived"), FUTURE);
- UserCredentials uc = new UserCredentials(Map.of("s3a", cred));
- String str = uc.toString();
- assertFalse(str.contains(TOKEN), "UserCredentials must not contain raw identity token");
- // ServiceCredential values should also be redacted in nested toString
- assertFalse(str.contains("short-lived"),
- "UserCredentials must not expose ServiceCredential property values");
- }
-
- @Test
- public void testUserCredentialsSerializationRoundTrip() throws Exception {
- Map creds = new HashMap<>();
- creds.put("s3a", new ServiceCredential(Map.of("key", "val1"), FUTURE));
- creds.put("abfss", new ServiceCredential(Map.of("token", "val2"), NOW));
- UserCredentials original = new UserCredentials(creds);
-
- byte[] bytes = serialize(original);
- UserCredentials deserialized = (UserCredentials) deserialize(bytes);
-
- assertEquals(original, deserialized);
- assertEquals(original.getCredentials(), deserialized.getCredentials());
- }
-
- @Test
- public void testUserCredentialsDefensiveCopy() {
- Map creds = new HashMap<>();
- creds.put("s3a", new ServiceCredential(Map.of("k", "v"), FUTURE));
- UserCredentials uc = new UserCredentials(creds);
- // Mutating the original map must not affect the credentials
- creds.put("extra", new ServiceCredential(Map.of("x", "y"), NOW));
- assertFalse(uc.forScheme("extra").isPresent());
- }
-
- @Test
- public void testUserCredentialsEqualsAndHashCode() {
- ServiceCredential cred = new ServiceCredential(Map.of("k", "v"), FUTURE);
- UserCredentials a = new UserCredentials(Map.of("s3a", cred));
- UserCredentials b = new UserCredentials(Map.of("s3a", cred));
- UserCredentials c = new UserCredentials(Map.of("abfss", cred));
- assertEquals(a, b);
- assertEquals(a.hashCode(), b.hashCode());
- assertNotEquals(a, c);
- }
-
- @Test
- public void testUserCredentialsForSchemeCaseInsensitive() {
- ServiceCredential cred = new ServiceCredential(Map.of("key", "val"), FUTURE);
- // Store with lowercase key, lookup with uppercase
- UserCredentials uc1 = new UserCredentials(Map.of("s3a", cred));
- assertTrue(uc1.forScheme("S3A").isPresent(), "Uppercase lookup should find lowercase key");
- assertEquals(cred, uc1.forScheme("S3A").get());
- assertTrue(uc1.forScheme("s3a").isPresent(), "Exact case lookup should still work");
- assertTrue(uc1.forScheme("S3a").isPresent(), "Mixed case lookup should work");
-
- // Store with uppercase key, lookup with lowercase
- UserCredentials uc2 = new UserCredentials(Map.of("ABFSS", cred));
- assertTrue(uc2.forScheme("abfss").isPresent(), "Lowercase lookup should find uppercase key");
- assertEquals(cred, uc2.forScheme("abfss").get());
-
- // Absent scheme still returns empty
- assertFalse(uc2.forScheme("hdfs").isPresent(), "Absent scheme should return empty");
- }
-
- @Test
- public void testUserCredentialsRejectsCaseCollidingSchemes() {
- Map creds = new HashMap<>();
- creds.put("s3a", new ServiceCredential(Map.of("k", "v1"), FUTURE));
- creds.put("S3A", new ServiceCredential(Map.of("k", "v2"), FUTURE));
- IllegalArgumentException e = assertThrows(IllegalArgumentException.class,
- () -> new UserCredentials(creds));
- assertTrue(e.getMessage().contains("Duplicate scheme"),
- "Exception message should identify the collision");
- }
-
- @Test
- public void testUserCredentialsRejectsNullSchemeKey() {
- Map creds = new HashMap<>();
- creds.put(null, new ServiceCredential(Map.of("k", "v"), FUTURE));
- assertThrows(NullPointerException.class, () -> new UserCredentials(creds));
- }
-
- // --- Helpers ---
-
- private static byte[] serialize(Object obj) throws Exception {
- ByteArrayOutputStream baos = new ByteArrayOutputStream();
- try (ObjectOutputStream oos = new ObjectOutputStream(baos)) {
- oos.writeObject(obj);
- }
- return baos.toByteArray();
- }
-
- private static Object deserialize(byte[] bytes) throws Exception {
- ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
- try (ObjectInputStream ois = new ObjectInputStream(bais)) {
- return ois.readObject();
- }
- }
-}
diff --git a/core/src/test/java/org/apache/spark/security/FakeCredentialProvider.java b/core/src/test/java/org/apache/spark/security/FakeCredentialProvider.java
deleted file mode 100644
index a733803576cf5..0000000000000
--- a/core/src/test/java/org/apache/spark/security/FakeCredentialProvider.java
+++ /dev/null
@@ -1,79 +0,0 @@
-/*
- * 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.security;
-
-import java.net.URI;
-import java.time.Instant;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.atomic.AtomicInteger;
-
-/**
- * A fake credential provider for testing. Supports schemes "fake" and "shared".
- */
-public class FakeCredentialProvider implements CredentialProvider {
-
- /** Sentinel URI host that triggers a CredentialResolutionException. */
- public static final String ERROR_HOST = "error.example.com";
-
- private volatile Map initConf;
- private final AtomicInteger initCount = new AtomicInteger();
- private final AtomicInteger closeCount = new AtomicInteger();
-
- @Override
- public void init(Map conf) {
- this.initConf = conf;
- this.initCount.incrementAndGet();
- }
-
- @Override
- public Set supportedSchemes() {
- return Set.of("fake", "shared");
- }
-
- @Override
- public ServiceCredential resolve(UserContext user, URI target)
- throws CredentialResolutionException {
- if (target.getHost() != null && target.getHost().equals(ERROR_HOST)) {
- throw new CredentialResolutionException(
- "Simulated failure for target: " + target);
- }
- Instant expiresAt = Instant.now().plus(suggestedTtl());
- return new ServiceCredential(Map.of("provider", "fake"), expiresAt);
- }
-
- @Override
- public void close() {
- this.closeCount.incrementAndGet();
- }
-
- /** Returns the configuration map passed to {@link #init(Map)}, or null if not yet called. */
- public Map getInitConf() {
- return initConf;
- }
-
- /** Returns the number of times {@link #init(Map)} has been called. */
- public int getInitCount() {
- return initCount.get();
- }
-
- /** Returns the number of times {@link #close()} has been called. */
- public int getCloseCount() {
- return closeCount.get();
- }
-}
diff --git a/core/src/test/java/org/apache/spark/security/FileTokenIngestorSuite.java b/core/src/test/java/org/apache/spark/security/FileTokenIngestorSuite.java
deleted file mode 100644
index 9929e23964b32..0000000000000
--- a/core/src/test/java/org/apache/spark/security/FileTokenIngestorSuite.java
+++ /dev/null
@@ -1,283 +0,0 @@
-/*
- * 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.security;
-
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.nio.file.attribute.FileTime;
-import java.security.KeyPair;
-import java.security.KeyPairGenerator;
-import java.util.Comparator;
-import java.util.Date;
-import java.util.Optional;
-
-import io.jsonwebtoken.Jwts;
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-
-import static org.junit.jupiter.api.Assertions.*;
-
-public class FileTokenIngestorSuite {
-
- private Path tempDir;
-
- @BeforeEach
- public void setUp() throws Exception {
- tempDir = Files.createTempDirectory("token-ingestor-test");
- }
-
- @AfterEach
- public void tearDown() throws Exception {
- if (tempDir != null) {
- Files.walk(tempDir).sorted(Comparator.reverseOrder())
- .forEach(path -> {
- try { Files.deleteIfExists(path); } catch (Exception e) { /* cleanup, ignore */ }
- });
- }
- }
-
- private String createUnsignedJwt(String subject, String issuer, Date issuedAt, Date expiresAt) {
- var builder = Jwts.builder()
- .subject(subject)
- .issuer(issuer);
- if (issuedAt != null) builder.issuedAt(issuedAt);
- if (expiresAt != null) builder.expiration(expiresAt);
- return builder.compact();
- }
-
- private String createUnsignedJwt(String subject, String issuer) {
- return createUnsignedJwt(subject, issuer,
- new Date(), new Date(System.currentTimeMillis() + 3600000));
- }
-
- private String createUnsignedJwt() {
- return createUnsignedJwt("system:serviceaccount:ns:sa",
- "https://kubernetes.default.svc");
- }
-
- private String createSignedJwt(String subject, String issuer) throws Exception {
- KeyPairGenerator kpg = KeyPairGenerator.getInstance("RSA");
- kpg.initialize(2048);
- KeyPair kp = kpg.generateKeyPair();
- return Jwts.builder()
- .subject(subject)
- .issuer(issuer)
- .issuedAt(new Date())
- .expiration(new Date(System.currentTimeMillis() + 3600000))
- .signWith(kp.getPrivate())
- .compact();
- }
-
- private void writeToken(Path path, String content) throws Exception {
- Files.write(path, content.getBytes("UTF-8"));
- }
-
- @Test
- public void loadReturnsValidUserContextFromTokenFile() throws Exception {
- Path tokenFile = tempDir.resolve("token");
- String token = createUnsignedJwt();
- writeToken(tokenFile, token);
-
- FileTokenIngestor ingestor = new FileTokenIngestor(tokenFile);
- Optional result = ingestor.load();
-
- assertTrue(result.isPresent());
- UserContext ctx = result.get();
- assertEquals("system:serviceaccount:ns:sa", ctx.getPrincipal());
- assertEquals("https://kubernetes.default.svc", ctx.getIssuer());
- assertEquals(token, ctx.getRawToken());
- assertNotNull(ctx.getIssuedAt());
- assertNotNull(ctx.getExpiresAt());
- }
-
- @Test
- public void loadWorksWithSignedJwt() throws Exception {
- Path tokenFile = tempDir.resolve("token");
- String token = createSignedJwt("user@corp.example.com", "https://oidc.eks.us-west-2.amazonaws.com/id/ABC123");
- writeToken(tokenFile, token);
-
- FileTokenIngestor ingestor = new FileTokenIngestor(tokenFile);
- Optional result = ingestor.load();
-
- assertTrue(result.isPresent());
- UserContext ctx = result.get();
- assertEquals("user@corp.example.com", ctx.getPrincipal());
- assertEquals("https://oidc.eks.us-west-2.amazonaws.com/id/ABC123", ctx.getIssuer());
- assertEquals(token, ctx.getRawToken());
- assertNotNull(ctx.getIssuedAt());
- assertNotNull(ctx.getExpiresAt());
- }
-
- @Test
- public void loadDetectsFileRotation() throws Exception {
- Path tokenFile = tempDir.resolve("token");
- String token1 = createUnsignedJwt("user1", "https://issuer.example.com");
- writeToken(tokenFile, token1);
-
- FileTokenIngestor ingestor = new FileTokenIngestor(tokenFile);
- Optional result1 = ingestor.load();
- assertEquals("user1", result1.get().getPrincipal());
-
- // Simulate token rotation: sleep to ensure mtime differs
- Thread.sleep(100);
- String token2 = createUnsignedJwt("user2", "https://issuer.example.com");
- writeToken(tokenFile, token2);
-
- Optional result2 = ingestor.load();
- assertEquals("user2", result2.get().getPrincipal());
- assertEquals(token2, result2.get().getRawToken());
- }
-
- @Test
- public void loadDetectsFileRotationWhenMtimeIsUnchanged() throws Exception {
- Path tokenFile = tempDir.resolve("token");
- String token1 = createUnsignedJwt("user1", "https://issuer.example.com");
- writeToken(tokenFile, token1);
-
- FileTokenIngestor ingestor = new FileTokenIngestor(tokenFile);
- assertEquals("user1", ingestor.load().get().getPrincipal());
- FileTime originalMtime = Files.getLastModifiedTime(tokenFile);
-
- String token2 = createUnsignedJwt("user2", "https://issuer.example.com");
- writeToken(tokenFile, token2);
- Files.setLastModifiedTime(tokenFile, originalMtime);
-
- Optional result = ingestor.load();
- assertEquals("user2", result.get().getPrincipal());
- assertEquals(token2, result.get().getRawToken());
- }
-
- @Test
- public void loadReturnsEmptyForMissingFile() {
- Path tokenFile = tempDir.resolve("nonexistent");
- FileTokenIngestor ingestor = new FileTokenIngestor(tokenFile);
- assertTrue(ingestor.load().isEmpty());
- }
-
- @Test
- public void loadReturnsEmptyForEmptyFile() throws Exception {
- Path tokenFile = tempDir.resolve("token");
- writeToken(tokenFile, "");
-
- FileTokenIngestor ingestor = new FileTokenIngestor(tokenFile);
- assertTrue(ingestor.load().isEmpty());
- }
-
- @Test
- public void loadReturnsEmptyForWhitespaceOnlyFile() throws Exception {
- Path tokenFile = tempDir.resolve("token");
- writeToken(tokenFile, " \n\t ");
-
- FileTokenIngestor ingestor = new FileTokenIngestor(tokenFile);
- assertTrue(ingestor.load().isEmpty());
- }
-
- @Test
- public void loadReturnsEmptyForMalformedJwt() throws Exception {
- Path tokenFile = tempDir.resolve("token");
- writeToken(tokenFile, "not-a-jwt");
-
- FileTokenIngestor ingestor = new FileTokenIngestor(tokenFile);
- assertTrue(ingestor.load().isEmpty());
- }
-
- @Test
- public void loadReturnsEmptyForJwtMissingSubClaim() throws Exception {
- Path tokenFile = tempDir.resolve("token");
- String token = Jwts.builder()
- .issuer("https://issuer.example.com")
- .compact();
- writeToken(tokenFile, token);
-
- FileTokenIngestor ingestor = new FileTokenIngestor(tokenFile);
- assertTrue(ingestor.load().isEmpty());
- }
-
- @Test
- public void loadReturnsEmptyForJwtMissingIssClaim() throws Exception {
- Path tokenFile = tempDir.resolve("token");
- String token = Jwts.builder()
- .subject("user1")
- .compact();
- writeToken(tokenFile, token);
-
- FileTokenIngestor ingestor = new FileTokenIngestor(tokenFile);
- assertTrue(ingestor.load().isEmpty());
- }
-
- @Test
- public void loadHandlesJwtWithoutIatAndExpClaims() throws Exception {
- Path tokenFile = tempDir.resolve("token");
- String token = Jwts.builder()
- .subject("user1")
- .issuer("https://issuer.example.com")
- .compact();
- writeToken(tokenFile, token);
-
- FileTokenIngestor ingestor = new FileTokenIngestor(tokenFile);
- Optional result = ingestor.load();
- assertTrue(result.isPresent());
- assertEquals("user1", result.get().getPrincipal());
- assertNull(result.get().getIssuedAt());
- assertNull(result.get().getExpiresAt());
- }
-
- @Test
- public void loadUsesCachedResultWhenFileUnchanged() throws Exception {
- Path tokenFile = tempDir.resolve("token");
- String token = createUnsignedJwt("cached-user", "https://issuer.example.com");
- writeToken(tokenFile, token);
-
- FileTokenIngestor ingestor = new FileTokenIngestor(tokenFile);
- Optional result1 = ingestor.load();
- Optional result2 = ingestor.load();
-
- // Reference identity proves the cached path was hit (not re-parsed)
- assertSame(result1.get(), result2.get());
- assertEquals("cached-user", result1.get().getPrincipal());
- }
-
- @Test
- public void loadWorksWithKubernetesProjectedToken() throws Exception {
- Path tokenFile = tempDir.resolve("token");
- String token = createUnsignedJwt(
- "system:serviceaccount:spark-ns:spark-driver-sa",
- "https://oidc.eks.us-west-2.amazonaws.com/id/EXAMPLED539D4633E53DE1B716D3041E");
- writeToken(tokenFile, token);
-
- FileTokenIngestor ingestor = new FileTokenIngestor(tokenFile);
- Optional result = ingestor.load();
- assertTrue(result.isPresent());
- assertEquals("system:serviceaccount:spark-ns:spark-driver-sa", result.get().getPrincipal());
- assertTrue(result.get().getIssuer().contains("oidc.eks"));
- }
-
- @Test
- public void loadWorksWithPerUserIdentityToken() throws Exception {
- Path tokenFile = tempDir.resolve("token");
- String token = createUnsignedJwt("user@example.com", "https://accounts.google.com");
- writeToken(tokenFile, token);
-
- FileTokenIngestor ingestor = new FileTokenIngestor(tokenFile);
- Optional result = ingestor.load();
- assertTrue(result.isPresent());
- assertEquals("user@example.com", result.get().getPrincipal());
- assertEquals("https://accounts.google.com", result.get().getIssuer());
- }
-}
diff --git a/core/src/test/resources/META-INF/services/org.apache.spark.security.CredentialProvider b/core/src/test/resources/META-INF/services/org.apache.spark.security.CredentialProvider
deleted file mode 100644
index be245cc5429a6..0000000000000
--- a/core/src/test/resources/META-INF/services/org.apache.spark.security.CredentialProvider
+++ /dev/null
@@ -1,19 +0,0 @@
-#
-# 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.
-#
-
-org.apache.spark.security.FakeCredentialProvider
-org.apache.spark.security.AnotherFakeCredentialProvider
diff --git a/core/src/test/scala/org/apache/spark/deploy/security/UserCredentialManagerSuite.scala b/core/src/test/scala/org/apache/spark/deploy/security/UserCredentialManagerSuite.scala
deleted file mode 100644
index 5e881c8d3a3ef..0000000000000
--- a/core/src/test/scala/org/apache/spark/deploy/security/UserCredentialManagerSuite.scala
+++ /dev/null
@@ -1,506 +0,0 @@
-/*
- * 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.deploy.security
-
-import java.time.Instant
-import java.util
-import java.util.Optional
-import java.util.concurrent.atomic.{AtomicInteger, AtomicReference}
-
-import scala.concurrent.duration._
-
-import org.scalatest.concurrent.Eventually.{eventually, timeout}
-
-import org.apache.spark.{SparkConf, SparkFunSuite}
-import org.apache.spark.internal.config._
-import org.apache.spark.security._
-
-class UserCredentialManagerSuite extends SparkFunSuite {
-
- override def beforeEach(): Unit = {
- super.beforeEach()
- CredentialProviderLoader.resetForTesting()
- }
-
- private def createSparkConf(): SparkConf = {
- new SparkConf(loadDefaults = false)
- .set(SECURITY_OIDC_ENABLED, true)
- .set(SECURITY_OIDC_IDENTITY_TOKEN_FILE, "/tmp/fake-token")
- .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 5000L) // 5s for tests
- .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 1000L) // 1s for tests
- }
-
- private def createUserContext(
- expiresInSeconds: Long = 300): UserContext = {
- new UserContext(
- "test-user",
- "https://issuer.example.com",
- "fake.jwt.token",
- Instant.now(),
- Instant.now().plusSeconds(expiresInSeconds))
- }
-
- private def createIngestor(ctx: UserContext): TokenIngestor = {
- new TokenIngestor {
- override def load(): Optional[UserContext] = Optional.of(ctx)
- }
- }
-
- private def createFailingIngestor(): TokenIngestor = {
- new TokenIngestor {
- override def load(): Optional[UserContext] = Optional.empty()
- }
- }
-
- test("start() acquires initial credentials and invokes callback") {
- val conf = createSparkConf()
- val ctx = createUserContext()
- val callbackRef = new AtomicReference[Array[Byte]]()
-
- // Use CredentialProviderLoader with the FakeCredentialProvider from ServiceLoader
- // FakeCredentialProvider supports scheme "fake"
- conf.set("spark.security.oidc.provider.fake",
- "org.apache.spark.security.FakeCredentialProvider")
-
- val manager = new UserCredentialManager(
- conf,
- createIngestor(ctx),
- bytes => callbackRef.set(bytes))
-
- try {
- val result = manager.start()
- assert(result != null, "start() should return serialized credentials")
- assert(callbackRef.get() != null, "callback should have been invoked")
-
- // Verify deserialization
- val creds = UserCredentialManager.deserializeUserCredentials(result)
- assert(creds.forScheme("fake").isPresent,
- "Should have credentials for 'fake' scheme")
- assert(creds.forScheme("fake").get().getProperties.get("provider") === "fake")
- } finally {
- manager.stop()
- }
- }
-
- test("start() throws when TokenIngestor returns empty (fail-fast)") {
- val conf = createSparkConf()
- conf.set("spark.security.oidc.provider.fake",
- "org.apache.spark.security.FakeCredentialProvider")
-
- val manager = new UserCredentialManager(
- conf,
- createFailingIngestor(),
- _ => ())
-
- try {
- val ex = intercept[IllegalStateException] {
- manager.start()
- }
- assert(ex.getMessage.contains("identity token file is missing or malformed"))
- } finally {
- manager.stop()
- }
- }
-
- test("serialization round-trip of UserCredentials") {
- val props = new util.HashMap[String, String]()
- props.put("fs.s3a.access.key", "AKIAEXAMPLE")
- props.put("fs.s3a.secret.key", "secret123")
- props.put("fs.s3a.session.token", "token456")
- val cred = new ServiceCredential(props, Instant.now().plusSeconds(3600))
-
- val credsMap = new util.HashMap[String, ServiceCredential]()
- credsMap.put("s3a", cred)
- val original = new UserCredentials(credsMap)
-
- val conf = createSparkConf()
- val manager = new UserCredentialManager(conf, createFailingIngestor(), _ => ())
-
- try {
- val serialized = UserCredentialManager.serializeUserCredentials(original)
- val deserialized = UserCredentialManager.deserializeUserCredentials(serialized)
-
- assert(deserialized.forScheme("s3a").isPresent)
- val restored = deserialized.forScheme("s3a").get()
- assert(restored.getProperties.get("fs.s3a.access.key") === "AKIAEXAMPLE")
- assert(restored.getProperties.get("fs.s3a.secret.key") === "secret123")
- assert(restored.getProperties.get("fs.s3a.session.token") === "token456")
- assert(restored.getExpiresAt === cred.getExpiresAt)
- } finally {
- manager.stop()
- }
- }
-
- test("deserialization rejects unauthorized classes (ObjectInputFilter)") {
- // Serialize a class that is NOT in the allowed filter pattern.
- // The filter allows org.apache.spark.security.**, java.util.**, java.time.**,
- // java.lang.** but blocks everything else (including java.io.File).
- val bos = new java.io.ByteArrayOutputStream()
- val oos = new java.io.ObjectOutputStream(bos)
- oos.writeObject(new java.io.File("/tmp/malicious"))
- oos.close()
-
- val ex = intercept[java.io.InvalidClassException] {
- UserCredentialManager.deserializeUserCredentials(bos.toByteArray)
- }
- // ObjectInputFilter rejects classes not in the allowlist
- assert(ex.getMessage.contains("REJECTED") || ex.getMessage.contains("filter"),
- s"Expected filter rejection, got: ${ex.getMessage}")
- }
-
- test("computeRenewalDelay respects safetyMargin and minInterval") {
- val conf = createSparkConf()
- .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 10000L) // 10s
- .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 5000L) // 5s
-
- val manager = new UserCredentialManager(conf, createFailingIngestor(), _ => ())
- try {
- // Token expires in 60s, credential expires in 30s
- // Expected: min(60s, 30s) - 10s = 20s
- val ctx = createUserContext(expiresInSeconds = 60)
- val credExpiry = Some(Instant.now().plusSeconds(30))
- val delay = manager.computeRenewalDelay(ctx, credExpiry)
-
- // Allow 1s tolerance for timing
- assert(delay >= 19000 && delay <= 21000,
- s"Expected ~20000ms, got ${delay}ms")
- } finally {
- manager.stop()
- }
- }
-
- test("computeRenewalDelay uses minInterval when expiry is very close") {
- val conf = createSparkConf()
- .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 10000L)
- .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 5000L)
-
- val manager = new UserCredentialManager(conf, createFailingIngestor(), _ => ())
- try {
- // Token expires in 5s, safetyMargin is 10s -> computed delay would be negative
- // Should be bounded by minInterval (5s)
- val ctx = createUserContext(expiresInSeconds = 5)
- val credExpiry = Some(Instant.now().plusSeconds(5))
- val delay = manager.computeRenewalDelay(ctx, credExpiry)
-
- assert(delay === 5000L, s"Expected minInterval (5000ms), got ${delay}ms")
- } finally {
- manager.stop()
- }
- }
-
- test("computeRenewalDelay uses identity token expiry when credential has no expiry") {
- val conf = createSparkConf()
- .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 10000L) // 10s
- .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 5000L) // 5s
-
- val manager = new UserCredentialManager(conf, createFailingIngestor(), _ => ())
- try {
- // Token expires in 60s, no credential expiry
- // Expected: 60s - 10s = 50s
- val ctx = createUserContext(expiresInSeconds = 60)
- val delay = manager.computeRenewalDelay(ctx, None)
-
- assert(delay >= 49000 && delay <= 51000,
- s"Expected ~50000ms, got ${delay}ms")
- } finally {
- manager.stop()
- }
- }
-
- test("computeRenewalDelay returns default when no expiry information") {
- val conf = createSparkConf()
- val manager = new UserCredentialManager(conf, createFailingIngestor(), _ => ())
- try {
- // UserContext with null expiresAt
- val ctx = new UserContext(
- "test-user", "https://issuer.example.com", "fake.jwt.token",
- Instant.now(), null)
- val delay = manager.computeRenewalDelay(ctx, None)
-
- // Should be DEFAULT_RENEWAL_INTERVAL_NO_EXPIRY_MS = 7 minutes = 420000ms
- assert(delay === 420000L, s"Expected 420000ms (7 min), got ${delay}ms")
- } finally {
- manager.stop()
- }
- }
-
- test("computeBackoffDelay increases exponentially") {
- val conf = createSparkConf()
- .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 1000L)
-
- val manager = new UserCredentialManager(conf, createFailingIngestor(), _ => ())
- try {
- val failuresField = classOf[UserCredentialManager].getDeclaredField("consecutiveFailures")
- failuresField.setAccessible(true)
-
- // First failure: base = minInterval * 2^0 = 1000ms
- failuresField.setInt(manager, 1)
- val delay1 = manager.computeBackoffDelay()
- assert(delay1 >= 1000 && delay1 <= 1200,
- s"First backoff should be ~1000-1100ms, got ${delay1}ms")
-
- // Second failure: base = minInterval * 2^1 = 2000ms
- failuresField.setInt(manager, 2)
- val delay2 = manager.computeBackoffDelay()
- assert(delay2 >= 2000 && delay2 <= 2300,
- s"Second backoff should be ~2000-2200ms, got ${delay2}ms")
-
- // Third failure: base = minInterval * 2^2 = 4000ms
- failuresField.setInt(manager, 3)
- val delay3 = manager.computeBackoffDelay()
- assert(delay3 >= 4000 && delay3 <= 4500,
- s"Third backoff should be ~4000-4400ms, got ${delay3}ms")
- } finally {
- manager.stop()
- }
- }
-
- test("computeBackoffDelay is capped at maxBackoffMs") {
- val conf = createSparkConf()
- .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 1000L)
-
- val manager = new UserCredentialManager(conf, createFailingIngestor(), _ => ())
- try {
- val failuresField = classOf[UserCredentialManager].getDeclaredField("consecutiveFailures")
- failuresField.setAccessible(true)
-
- // Many failures: should be capped at 10 minutes (600000ms)
- failuresField.setInt(manager, 20)
- val delay = manager.computeBackoffDelay()
- assert(delay <= 660000L, // 600000 + 10% jitter
- s"Backoff should be capped at ~600000ms + jitter, got ${delay}ms")
- } finally {
- manager.stop()
- }
- }
-
- test("computeBackoffDelay handles zero consecutiveFailures defensively") {
- val conf = createSparkConf()
- .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 1000L)
-
- val manager = new UserCredentialManager(conf, createFailingIngestor(), _ => ())
- try {
- val failuresField = classOf[UserCredentialManager].getDeclaredField("consecutiveFailures")
- failuresField.setAccessible(true)
-
- // Edge case: 0 failures (should not happen in practice, but defensive)
- failuresField.setInt(manager, 0)
- val delay = manager.computeBackoffDelay()
- // shiftAmount = max(0, min(0-1, 6)) = max(0, -1) = 0
- // baseDelay = 1000 * 2^0 = 1000
- assert(delay >= 1000 && delay <= 1200,
- s"With 0 failures, backoff should be ~1000ms, got ${delay}ms")
- } finally {
- manager.stop()
- }
- }
-
- test("UserCredentialManager.create returns None when disabled") {
- val conf = new SparkConf(loadDefaults = false)
- .set(SECURITY_OIDC_ENABLED, false)
-
- val result = UserCredentialManager.create(conf, _ => ())
- assert(result.isEmpty)
- }
-
- test("UserCredentialManager.create returns Some when enabled with valid config") {
- val conf = createSparkConf()
- val result = UserCredentialManager.create(conf, _ => ())
- assert(result.isDefined)
- }
-
- test("UserCredentialManager.create throws when enabled without token file") {
- val conf = new SparkConf(loadDefaults = false)
- .set(SECURITY_OIDC_ENABLED, true)
- // Deliberately not setting SECURITY_OIDC_IDENTITY_TOKEN_FILE
-
- val ex = intercept[IllegalArgumentException] {
- UserCredentialManager.create(conf, _ => ())
- }
- assert(ex.getMessage.contains("spark.security.oidc.identityToken.file"))
- }
-
- test("renewal is scheduled after successful credential acquisition") {
- val conf = createSparkConf()
- conf.set("spark.security.oidc.provider.fake",
- "org.apache.spark.security.FakeCredentialProvider")
- val ctx = createUserContext(expiresInSeconds = 60)
- var callbackCount = 0
-
- val manager = new UserCredentialManager(
- conf,
- createIngestor(ctx),
- _ => { callbackCount += 1 })
-
- try {
- val result = manager.start()
- assert(result != null)
- assert(callbackCount === 1, "callback should be invoked once on start")
- } finally {
- manager.stop()
- }
- }
-
- test("stop() after start() does not throw") {
- val conf = createSparkConf()
- conf.set("spark.security.oidc.provider.fake",
- "org.apache.spark.security.FakeCredentialProvider")
- val ctx = createUserContext()
-
- val manager = new UserCredentialManager(conf, createIngestor(ctx), _ => ())
- manager.start()
- // Should not throw
- manager.stop()
- }
-
- test("per-provider error isolation: one provider failure does not block others") {
- // Configure two schemes: "fake" (will succeed) and "shared" (will trigger ambiguity
- // because no explicit provider is set for it and both FakeCredentialProvider and
- // AnotherFakeCredentialProvider support "shared").
- // The per-provider catch should absorb the IllegalArgumentException for "shared"
- // and still return credentials from "fake".
- val conf = createSparkConf()
- // Only set explicit provider for "fake", leave "shared" ambiguous
- conf.set("spark.security.oidc.provider.fake",
- "org.apache.spark.security.FakeCredentialProvider")
- conf.set("spark.security.oidc.provider.shared", "nonexistent.Provider")
-
- val ctx = createUserContext()
- val callbackRef = new AtomicReference[Array[Byte]]()
-
- val manager = new UserCredentialManager(
- conf,
- createIngestor(ctx),
- bytes => callbackRef.set(bytes))
-
- try {
- val result = manager.start()
- assert(result != null, "start() should return serialized credentials")
-
- // Verify that "fake" credentials were resolved despite "shared" failing
- val creds = UserCredentialManager.deserializeUserCredentials(result)
- assert(creds.forScheme("fake").isPresent,
- "Should have credentials for 'fake' scheme even though 'shared' failed")
- assert(creds.forScheme("fake").get().getProperties.get("provider") === "fake")
- // "shared" should NOT be present (its provider was not found)
- assert(!creds.forScheme("shared").isPresent,
- "'shared' scheme should not have credentials due to provider resolution failure")
- } finally {
- manager.stop()
- }
- }
-
- test("per-provider error isolation: all providers fail throws IllegalStateException") {
- // Configure only "shared" with a non-existent provider class
- val conf = createSparkConf()
- conf.set("spark.security.oidc.provider.shared", "nonexistent.Provider")
-
- val ctx = createUserContext()
-
- val manager = new UserCredentialManager(conf, createIngestor(ctx), _ => ())
-
- try {
- val ex = intercept[IllegalStateException] {
- manager.start()
- }
- assert(ex.getMessage.contains("No credential providers resolved any credentials"))
- } finally {
- manager.stop()
- }
- }
-
- test("token rotation triggers new credential acquisition and propagation") {
- val conf = createSparkConf()
- .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 1000L) // 1s
- .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 500L) // 0.5s
- conf.set("spark.security.oidc.provider.fake",
- "org.apache.spark.security.FakeCredentialProvider")
-
- // First token: user-A
- val ctx1 = new UserContext(
- "user-A", "https://issuer.example.com", "token-A",
- Instant.now(), Instant.now().plusSeconds(2)) // expires in 2s
-
- // Second token: user-B (simulating rotation)
- val ctx2 = new UserContext(
- "user-B", "https://issuer.example.com", "token-B",
- Instant.now(), Instant.now().plusSeconds(300))
-
- // TokenIngestor that returns ctx1 initially, then ctx2 after first call
- val callCount = new AtomicInteger(0)
- val rotatingIngestor = new TokenIngestor {
- override def load(): Optional[UserContext] = {
- if (callCount.getAndIncrement() == 0) Optional.of(ctx1)
- else Optional.of(ctx2)
- }
- }
-
- val callbacks = new java.util.concurrent.CopyOnWriteArrayList[Array[Byte]]()
- val manager = new UserCredentialManager(
- conf,
- rotatingIngestor,
- bytes => callbacks.add(bytes))
-
- try {
- val initial = manager.start()
- assert(initial != null)
- assert(callbacks.size() === 1, "Should have one callback from start()")
-
- // Wait for renewal to fire (token expires in 2s, safety margin 1s -> renews after ~1s)
- eventually(timeout(10.seconds)) {
- // After renewal, a second callback should have been invoked with new credentials
- assert(callbacks.size() >= 2,
- s"Expected at least 2 callbacks (initial + renewal), got ${callbacks.size()}")
- }
-
- // Verify that the ingestor was called more than once (rotation detected)
- assert(callCount.get() >= 2,
- s"TokenIngestor should have been called at least twice, got ${callCount.get()}")
- } finally {
- manager.stop()
- }
- }
-
- test("stop() closes initialized credential providers") {
- val conf = createSparkConf()
- val ctx = createUserContext()
- val callbackRef = new AtomicReference[Array[Byte]]()
-
- conf.set("spark.security.oidc.provider.fake",
- "org.apache.spark.security.FakeCredentialProvider")
-
- val manager = new UserCredentialManager(
- conf,
- createIngestor(ctx),
- bytes => callbackRef.set(bytes))
-
- manager.start()
-
- // Get the FakeCredentialProvider instance to verify close was called
- val providerOpt = CredentialProviderLoader.providerFor("fake",
- new util.HashMap[String, String]())
- assert(providerOpt.isPresent)
- val fakeProvider = providerOpt.get().asInstanceOf[FakeCredentialProvider]
- assert(fakeProvider.getCloseCount === 0, "close() not yet called before stop()")
-
- manager.stop()
-
- assert(fakeProvider.getCloseCount === 1,
- "stop() should close initialized providers exactly once")
- }
-}
diff --git a/dev/create-release/release-build.sh b/dev/create-release/release-build.sh
index 63524f3ba7c14..161dca83738b3 100755
--- a/dev/create-release/release-build.sh
+++ b/dev/create-release/release-build.sh
@@ -683,7 +683,7 @@ SCALA_2_12_PROFILES="-Pscala-2.12"
HIVE_PROFILES="-Phive -Phive-thriftserver"
# Profiles for publishing snapshots and release to Maven Central
# We use Apache Hive 2.3 for publishing
-PUBLISH_PROFILES="$BASE_PROFILES $HIVE_PROFILES -Pspark-ganglia-lgpl -Pkinesis-asl -Pcredential-aws -Phadoop-cloud -Pjvm-profiler"
+PUBLISH_PROFILES="$BASE_PROFILES $HIVE_PROFILES -Pspark-ganglia-lgpl -Pkinesis-asl -Phadoop-cloud -Pjvm-profiler"
# Profiles for building binary releases
BASE_RELEASE_PROFILES="$BASE_PROFILES -Psparkr"
diff --git a/dev/lint-java b/dev/lint-java
index 8095de68fb2a3..ff431301773f3 100755
--- a/dev/lint-java
+++ b/dev/lint-java
@@ -20,7 +20,7 @@
SCRIPT_DIR="$( cd "$( dirname "$0" )" && pwd )"
SPARK_ROOT_DIR="$(dirname $SCRIPT_DIR)"
-ERRORS=$($SCRIPT_DIR/../build/mvn -Pkinesis-asl -Pcredential-aws -Pspark-ganglia-lgpl -Pkubernetes -Pyarn -Phive -Phive-thriftserver checkstyle:check | grep ERROR)
+ERRORS=$($SCRIPT_DIR/../build/mvn -Pkinesis-asl -Pspark-ganglia-lgpl -Pkubernetes -Pyarn -Phive -Phive-thriftserver checkstyle:check | grep ERROR)
if test ! -z "$ERRORS"; then
echo -e "Checkstyle checks failed at following occurrences:\n$ERRORS"
diff --git a/dev/mima b/dev/mima
index eecf1ad1081db..39be02a1dd557 100755
--- a/dev/mima
+++ b/dev/mima
@@ -24,7 +24,7 @@ set -e
FWDIR="$(cd "`dirname "$0"`"/..; pwd)"
cd "$FWDIR"
-SPARK_PROFILES=${1:-"-Pkubernetes -Pyarn -Pspark-ganglia-lgpl -Pkinesis-asl -Pcredential-aws -Phive-thriftserver -Phive"}
+SPARK_PROFILES=${1:-"-Pkubernetes -Pyarn -Pspark-ganglia-lgpl -Pkinesis-asl -Phive-thriftserver -Phive"}
# Capture sbt output to show errors if the command fails
OLD_DEPS_OUTPUT="$(build/sbt -DcopyDependencies=false $SPARK_PROFILES "export oldDeps/fullClasspath" 2>&1)" || {
diff --git a/dev/sbt-checkstyle b/dev/sbt-checkstyle
index a148af6039181..f2d5a0fa304ac 100755
--- a/dev/sbt-checkstyle
+++ b/dev/sbt-checkstyle
@@ -17,7 +17,7 @@
# limitations under the License.
#
-SPARK_PROFILES=${1:-"-Pkinesis-asl -Pcredential-aws -Pspark-ganglia-lgpl -Pkubernetes -Pyarn -Phive -Phive-thriftserver -Pjvm-profiler"}
+SPARK_PROFILES=${1:-"-Pkinesis-asl -Pspark-ganglia-lgpl -Pkubernetes -Pyarn -Phive -Phive-thriftserver -Pjvm-profiler"}
# NOTE: echo "q" is needed because SBT prompts the user for input on encountering a build file
# with failure (either resolution or compilation); the "q" makes SBT quit.
diff --git a/dev/scalastyle b/dev/scalastyle
index f77fb3e1ee224..09e6c2372614d 100755
--- a/dev/scalastyle
+++ b/dev/scalastyle
@@ -17,7 +17,7 @@
# limitations under the License.
#
-SPARK_PROFILES=${1:-"-Pkubernetes -Pyarn -Pspark-ganglia-lgpl -Pkinesis-asl -Pcredential-aws -Phive-thriftserver -Phive -Pvolcano -Pjvm-profiler -Phadoop-cloud -Pdocker-integration-tests -Pkubernetes-integration-tests"}
+SPARK_PROFILES=${1:-"-Pkubernetes -Pyarn -Pspark-ganglia-lgpl -Pkinesis-asl -Phive-thriftserver -Phive -Pvolcano -Pjvm-profiler -Phadoop-cloud -Pdocker-integration-tests -Pkubernetes-integration-tests"}
# NOTE: echo "q" is needed because SBT prompts the user for input on encountering a build file
# with failure (either resolution or compilation); the "q" makes SBT quit.
diff --git a/dev/spark-test-image-util/docs/build-docs b/dev/spark-test-image-util/docs/build-docs
index c2ee826832ef6..ca59769f24231 100755
--- a/dev/spark-test-image-util/docs/build-docs
+++ b/dev/spark-test-image-util/docs/build-docs
@@ -35,7 +35,7 @@ FWDIR="$(cd "`dirname "${BASH_SOURCE[0]}"`"; pwd)"
SPARK_HOME="$(cd "`dirname "${BASH_SOURCE[0]}"`"/../../..; pwd)"
# 1.Compile spark outside the container to prepare for generating documents inside the container.
-build/sbt -Phive -Pkinesis-asl -Pcredential-aws clean unidoc package
+build/sbt -Phive -Pkinesis-asl clean unidoc package
# 2.Build container image.
docker buildx build \
diff --git a/dev/sparktestsupport/modules.py b/dev/sparktestsupport/modules.py
index 79e2e5a7bc2e5..30d6299552955 100644
--- a/dev/sparktestsupport/modules.py
+++ b/dev/sparktestsupport/modules.py
@@ -410,21 +410,6 @@ def __hash__(self):
)
-credential_aws = Module(
- name="credential-aws",
- dependencies=[tags, core],
- source_file_regexes=[
- "connector/credential-aws/",
- ],
- build_profile_flags=[
- "-Pcredential-aws",
- ],
- sbt_test_goals=[
- "credential-aws/test",
- ],
-)
-
-
streaming_kafka_0_10 = Module(
name="streaming-kafka-0-10",
dependencies=[streaming, core],
diff --git a/dev/sparktestsupport/utils.py b/dev/sparktestsupport/utils.py
index 1657aa56da67f..029b1627bd0bd 100755
--- a/dev/sparktestsupport/utils.py
+++ b/dev/sparktestsupport/utils.py
@@ -121,8 +121,7 @@ def determine_modules_to_test(changed_modules, deduplicated=True):
>>> sorted([x.name for x in determine_modules_to_test(
... [modules.sql, modules.core], deduplicated=False)])
... # doctest: +NORMALIZE_WHITESPACE
- ['avro', 'catalyst', 'connect', 'core', 'credential-aws', 'docker-integration-tests',
- 'examples', 'graphx',
+ ['avro', 'catalyst', 'connect', 'core', 'docker-integration-tests', 'examples', 'graphx',
'hive', 'hive-thriftserver', 'mllib', 'mllib-local', 'pipelines', 'protobuf',
'pyspark-connect', 'pyspark-core', 'pyspark-errors', 'pyspark-ml', 'pyspark-ml-connect',
'pyspark-mllib', 'pyspark-pandas', 'pyspark-pandas-connect', 'pyspark-pandas-slow',
diff --git a/dev/test-dependencies.sh b/dev/test-dependencies.sh
index ad93304e73732..68c61232ea2af 100755
--- a/dev/test-dependencies.sh
+++ b/dev/test-dependencies.sh
@@ -31,7 +31,7 @@ export LC_ALL=C
# NOTE: These should match those in the release publishing script, and be kept in sync with
# dev/create-release/release-build.sh
HADOOP_MODULE_PROFILES="-Phive-thriftserver -Pkubernetes -Pyarn -Phive \
- -Pspark-ganglia-lgpl -Pkinesis-asl -Pcredential-aws -Phadoop-cloud -Pjvm-profiler"
+ -Pspark-ganglia-lgpl -Pkinesis-asl -Phadoop-cloud -Pjvm-profiler"
MVN="build/mvn"
HADOOP_HIVE_PROFILES=(
hadoop-3-hive-2.3
diff --git a/docs/_plugins/build_api_docs.rb b/docs/_plugins/build_api_docs.rb
index 4b6251b804237..429cef5aa026c 100644
--- a/docs/_plugins/build_api_docs.rb
+++ b/docs/_plugins/build_api_docs.rb
@@ -45,7 +45,7 @@ def build_spark_if_necessary
print_header "Building Spark."
cd(SPARK_PROJECT_ROOT)
- command = "NO_PROVIDED_SPARK_JARS=0 build/sbt -Phive -Pkinesis-asl -Pcredential-aws clean package"
+ command = "NO_PROVIDED_SPARK_JARS=0 build/sbt -Phive -Pkinesis-asl clean package"
puts "Running '#{command}'; this may take a few minutes..."
system(command) || raise("Failed to build Spark")
# SPARK-53327: Use the modified ResourceImpl.class in spark-catalyst which is compatible with Java 25
@@ -129,7 +129,7 @@ def build_spark_scala_and_java_docs_if_necessary
return
end
- command = "build/sbt -Pkinesis-asl -Pcredential-aws unidoc"
+ command = "build/sbt -Pkinesis-asl unidoc"
puts "Running '#{command}'..."
# Two filter passes on the unidoc output, plus an additive fatal-error summary:
diff --git a/pom.xml b/pom.xml
index a40658107ffb3..0753674f6f84e 100644
--- a/pom.xml
+++ b/pom.xml
@@ -3524,13 +3524,6 @@
-
- credential-aws
-
- connector/credential-aws
-
-
-
test-java-home
diff --git a/project/SparkBuild.scala b/project/SparkBuild.scala
index 4421bb334d86f..9e301a79f5476 100644
--- a/project/SparkBuild.scala
+++ b/project/SparkBuild.scala
@@ -72,10 +72,10 @@ object BuildCommons {
udfWorkerProjects
val optionallyEnabledProjects@Seq(kubernetes, yarn,
- sparkGangliaLgpl, streamingKinesisAsl, profiler, credentialAws,
+ sparkGangliaLgpl, streamingKinesisAsl, profiler,
dockerIntegrationTests, hadoopCloud, kubernetesIntegrationTests) =
Seq("kubernetes", "yarn",
- "ganglia-lgpl", "streaming-kinesis-asl", "profiler", "credential-aws",
+ "ganglia-lgpl", "streaming-kinesis-asl", "profiler",
"docker-integration-tests", "hadoop-cloud", "kubernetes-integration-tests").map(ProjectRef(buildLocation, _))
val assemblyProjects@Seq(networkYarn, streamingKafka010Assembly, streamingKinesisAslAssembly) =
@@ -416,8 +416,7 @@ object SparkBuild extends PomBuild {
Seq(
spark, hive, hiveThriftServer, repl, networkCommon, networkShuffle, networkYarn,
unsafe, tags, tokenProviderKafka010, sqlKafka010, pipelines, connectCommon, connect,
- connectJdbc, connectClient, variant, connectShims, profiler, credentialAws,
- commonUtilsJava, sparkConfig,
+ connectJdbc, connectClient, variant, connectShims, profiler, commonUtilsJava, sparkConfig,
udfWorkerProto, udfWorkerCore, udfWorkerGrpc
).contains(x)
}