[SPARK-57891][CORE] Add CredentialProvider SPI and ServiceLoader discovery

core/src/main/java/org/apache/spark/security/CredentialProvider.java

@@ -0,0 +1,93 @@
+/*
+ * 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.util.Map;
+import java.util.Set;
+
+import org.apache.spark.annotation.DeveloperApi;
+
+/**
+ * :: DeveloperApi ::
+ * Service Provider Interface for credential resolution in the OIDC credential propagation
+ * framework.
+ * <p>
+ * Implementations exchange a user's identity (represented by {@link UserContext}) for a
+ * short-lived {@link ServiceCredential} scoped to a target URI. Providers are discovered
+ * via {@link java.util.ServiceLoader} and selected based on the URI scheme.
+ *
+ * @since 4.3.0
+ */
+@DeveloperApi
+public interface CredentialProvider {
+
+  /**
+   * Initializes this provider with configuration properties.
+   * <p>
+   * Called exactly once per provider instance by {@link CredentialProviderLoader}
+   * (first-conf-wins semantics). Subsequent resolutions reuse the already-initialized
+   * instance without re-calling this method. Implementations should capture any configuration
+   * they need (e.g., endpoint URLs, role ARNs) from the provided map.
+   *
+   * @param conf Spark configuration properties as a string map (must not be null)
+   * @since 4.3.0
+   */
+  void init(Map<String, String> conf);
+
+  /**
+   * Returns the set of URI schemes this provider supports (e.g., {@code {"s3a"}}).
+   * <p>
+   * Scheme values are compared case-insensitively (normalized to lowercase). The returned
+   * set must be non-empty and stable across calls.
+   *
+   * @return a non-empty set of supported scheme names
+   * @since 4.3.0
+   */
+  Set<String> supportedSchemes();
+
+  /**
+   * Exchanges the user's identity for a short-lived service credential scoped to the
+   * given target URI.
+   * <p>
+   * 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.
+   * <p>
+   * 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);
+  }
+}

core/src/main/java/org/apache/spark/security/CredentialProviderLoader.java

@@ -0,0 +1,218 @@
+/*
+ * 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.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 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).
+ * <p>
+ * 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}.
+ * <p>
+ * When the configuration key {@code spark.security.credentials.provider.<scheme>} 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()}.
+ * <p>
+ * <b>Thread-safety:</b> This class uses synchronized access to the cached provider list and
+ * initialization tracking. Callers may invoke {@link #providerFor(String, Map)} from multiple
+ * threads. However, the returned {@link CredentialProvider} instances are not guaranteed to be
+ * thread-safe; callers should synchronize on the provider or confine it to a single thread.
+ *
+ * @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.credentials.provider.";
+
+  private static volatile List<CredentialProvider> 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<CredentialProvider> initializedProviders =
+      Collections.newSetFromMap(new IdentityHashMap<>());
+
+  private CredentialProviderLoader() {
+    // utility class
+  }
+
+  /**
+   * Returns the {@link CredentialProvider} for the given URI scheme, applying Binding Policy A:
+   * <ol>
+   *   <li>If no candidate supports the scheme, {@link Optional#empty()} is returned.</li>
+   *   <li>If {@code spark.security.credentials.provider.<scheme>} 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.</li>
+   *   <li>If unset and exactly one candidate supports the scheme, that candidate is selected.</li>
+   *   <li>If unset and multiple candidates support the scheme, an
+   *       {@link IllegalArgumentException} is thrown listing the candidates.</li>
+   * </ol>
+   * 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}.
+   * <p>
+   * Spark-internal callers pass their configuration as a {@code Map<String, String>} 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<CredentialProvider> providerFor(String scheme, Map<String, String> conf) {
+    Objects.requireNonNull(scheme, "scheme must not be null");
+    Objects.requireNonNull(conf, "conf must not be null");
+    String normalizedScheme = scheme.toLowerCase(Locale.ROOT);
+    List<CredentialProvider> providers = getProviders();
+
+    List<CredentialProvider> candidates = providers.stream()
+        .filter(p -> {
+          Set<String> 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).
+    synchronized (CredentialProviderLoader.class) {
+      if (!initializedProviders.contains(selected)) {
+        selected.init(conf);
+        initializedProviders.add(selected);
+      }
+    }
+    return Optional.of(selected);
+  }
+
+  /**
+   * Returns the cached list of discovered providers, loading them on first access.
+   */
+  private static List<CredentialProvider> getProviders() {
+    List<CredentialProvider> providers = cachedProviders;
+    if (providers == null) {
+      synchronized (CredentialProviderLoader.class) {
+        providers = cachedProviders;
+        if (providers == null) {
+          providers = loadProviders();
+          cachedProviders = providers;
+        }
+      }
+    }
+    return providers;
+  }
+
+  private static List<CredentialProvider> loadProviders() {
+    ClassLoader cl = Thread.currentThread().getContextClassLoader();
+    if (cl == null) {
+      cl = CredentialProvider.class.getClassLoader();
+    }
+    ServiceLoader<CredentialProvider> loader = ServiceLoader.load(CredentialProvider.class, cl);
+    List<CredentialProvider> result = new ArrayList<>();
+    for (CredentialProvider provider : loader) {
+      result.add(provider);
+    }
+    return result;
+  }
+
+  /**
+   * Resets the cached provider list and initialization tracking. Intended for testing only.
+   */
+  static void resetForTesting() {
+    synchronized (CredentialProviderLoader.class) {
+      cachedProviders = null;
+      initializedProviders.clear();
+    }
+  }
+
+  /**
+   * Overrides the cached provider list for testing. Intended for testing only.
+   */
+  static void setProvidersForTesting(List<CredentialProvider> providers) {
+    synchronized (CredentialProviderLoader.class) {
+      cachedProviders = providers;
+      initializedProviders.clear();
+    }
+  }
+}

core/src/main/java/org/apache/spark/security/CredentialResolutionException.java

@@ -0,0 +1,54 @@
+/*
+ * 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.
+ * <p>
+ * 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);
+  }
+}