sarutak commented on code in PR #57191: URL: https://github.com/apache/spark/pull/57191#discussion_r3570628277
########## 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); Review Comment: I have a security concern here. The full Spark configuration map is passed to `init()`. Since providers are discovered via ServiceLoader (potentially third-party JARs), a buggy or malicious provider receives all config properties including secrets from other subsystems (e.g., `spark.authenticate.secret`, database passwords, cloud keys). Spark has precedent for namespace scoping. [DataSourceV2Utils.extractSessionConfigs()](https://github.com/apache/spark/blob/9d760a91baa140fcf517a6b754ba75df3eab0f1f/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Utils.scala#L64) filters the full conf to only `spark.datasource.<keyPrefix>.*` before passing to providers. Similar to this approach, can we filter the map to only pass `spark.security.credentials.*` keys? ########## 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); Review Comment: What happens if `init()` throws? Currently the provider is not added to `initializedProviders`, so a subsequent call would retry `init()`. This retry behavior seems reasonable, but: - The provider instance may be in a half-initialized state after a failed `init()`. Is that safe to retry? - Should the contract document whether `init()` is expected to be idempotent-safe (i.e., safe to call again after a previous failure)? How about: 1. Documenting the retry-on-failure behavior in the `init()` Javadoc (e.g., "If init() throws, it may be retried on the next resolution attempt. Implementations should be safe to call again after a prior failure."), or 2. Adding a test verifying this behavior ########## 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; Review Comment: Since `CredentialProviderLoader` caches and reuses provider instances (singleton), `resolve()` may be called from multiple threads (e.g., credential refresh scheduling). Can we document whether implementations must be thread-safe ? (interface-level or `resolve()` Javadoc) Suggestion for interface-level Javadoc: ``` Implementations must be thread-safe: {@code resolve()} may be called concurrently from multiple threads after {@code init()} completes. ``` ########## 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"); Review Comment: Null scheme is properly rejected, but an empty string `""` passes through and silently returns `Optional.empty()`. An empty scheme is never valid per URI semantics (RFC 3986) and is likely always a caller bug. Can we add the following check? ```java if (scheme.isEmpty()) { throw new IllegalArgumentException("scheme must not be empty"); } ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
