You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "yadavay-amzn (via GitHub)" <gi...@apache.org> on 2026/07/30 17:55:26 UTC
[PR] [SPARK-57898][CORE] Implement AwsStsCredentialProvider (OIDC to STS credential exchange) [spark]
yadavay-amzn opened a new pull request, #57655:
URL: https://github.com/apache/spark/pull/57655
### What changes were proposed in this pull request?
Adds a new opt-in connector module `connector/credential-aws` containing `AwsStsCredentialProvider`, which implements the `CredentialProvider` SPI (SPARK-57891) to exchange an OIDC identity token for temporary AWS credentials via STS `AssumeRoleWithWebIdentity`. It returns S3A-compatible credentials (`fs.s3a.access.key` / `fs.s3a.secret.key` / `fs.s3a.session.token`) as a `ServiceCredential`.
- Configuration under `spark.security.oidc.*`: `roleArn` (required), `sessionName`, `durationSeconds`, `region`, `stsEndpoint`.
- Supports non-AWS STS-compatible endpoints (MinIO, Ceph, LocalStack) via `stsEndpoint`.
- Uses an anonymous STS client (web-identity needs no base AWS credentials).
- Thread-safe: configuration is resolved once in `init()` and published via a single `volatile` immutable holder read by `resolve()`.
- Never exposes the OIDC raw token in logs or exception messages.
This is the driver-side provider only (OIDC SPIP task 9). The executor-side `AwsCredentialsProvider` is tracked separately as SPARK-57899 (task 10).
### Why are the changes needed?
Part of the OIDC Credential Propagation SPIP ([SPARK-57703](https://issues.apache.org/jira/browse/SPARK-57703)). No `CredentialProvider` existed to exchange an OIDC token for AWS STS temporary credentials; this is the S3/STS reference provider.
### Does this PR introduce any user-facing change?
No change to defaults (`spark.security.oidc.enabled` remains off). Adds a new optional module and provider plus the `spark.security.oidc.*` AWS configuration keys.
### How was this patch tested?
New `AwsStsCredentialProviderSuite` (JUnit 5 + Mockito, 21 tests) with a mocked `StsClient`:
- Successful resolve with `ArgumentCaptor` assertions on the STS request (role ARN, web identity token, session name, duration) and on the mapped `fs.s3a.*` keys + expiration.
- STS failure wrapped in `CredentialResolutionException` (asserts the token never appears in the message chain).
- Incomplete STS credentials (null credentials / missing session token) throw `CredentialResolutionException` rather than NPE.
- Missing/blank `roleArn`, null/blank token, `resolve()` before `init()`.
- Resolved region/endpoint for endpoint+region, endpoint-only (default region), and neither.
- Session-name derivation, duration handling, `suggestedTtl`, `supportedSchemes`, and `ServiceLoader` discovery.
### Was this patch authored or co-authored using generative AI tooling?
Yes. Authored with assistance by Claude Opus 5.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57898][CORE] Implement AwsStsCredentialProvider (OIDC to STS credential exchange) [spark]
Posted by "yadavay-amzn (via GitHub)" <gi...@apache.org>.
yadavay-amzn commented on PR #57655:
URL: https://github.com/apache/spark/pull/57655#issuecomment-5247489331
Thanks for the review and approval @dongjoon-hyun. I just pushed the should-fix so the `resolve()` `IllegalStateException` catch is now genuinely exercised (added a mock `verify`). For the optional items, I'll take them in a follow-up together with the other follow-up items @sarutak flagged, to keep this PR focused. Thanks again!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57898][CORE] Implement AwsStsCredentialProvider (OIDC to STS credential exchange) [spark]
Posted by "sarutak (via GitHub)" <gi...@apache.org>.
sarutak commented on PR #57655:
URL: https://github.com/apache/spark/pull/57655#issuecomment-5266180631
**Merge Summary:**
- merged into master https://github.com/apache/spark/commit/3f598f070f91b1175f405f07f5aec427a8221bc5
- merged into branch-4.x https://github.com/apache/spark/commit/c39f1c6578b5b45d53e375ab7b7524aab92dee76
*Posted by `merge_spark_pr.py`*
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57898][CORE] Implement AwsStsCredentialProvider (OIDC to STS credential exchange) [spark]
Posted by "sarutak (via GitHub)" <gi...@apache.org>.
sarutak commented on PR #57655:
URL: https://github.com/apache/spark/pull/57655#issuecomment-5218723003
Hi @yadavay-amzn, the CI failure seems relevant to this change. Could you fix it?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57898][CORE] Implement AwsStsCredentialProvider (OIDC to STS credential exchange) [spark]
Posted by "sarutak (via GitHub)" <gi...@apache.org>.
sarutak commented on PR #57655:
URL: https://github.com/apache/spark/pull/57655#issuecomment-5259975675
Thank you for the thorough work and patience through all the review rounds so far, @yadavay-amzn.
Regarding `@VisibleForTesting`, I'm OK to replace it with a `// Visible for testing.` comment as this module doesn't use guava for anything else for now.
BTW, you say "No" in the `Was this patch authored or co-authored using generative AI tooling?` section in the description. Please let me confirm if this is true. If it's incorrect, please correct it.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57898][CORE] Implement AwsStsCredentialProvider (OIDC to STS credential exchange) [spark]
Posted by "sarutak (via GitHub)" <gi...@apache.org>.
sarutak commented on code in PR #57655:
URL: https://github.com/apache/spark/pull/57655#discussion_r3736717326
##########
connector/credential-aws/src/main/java/org/apache/spark/security/aws/AwsStsCredentialProvider.java:
##########
@@ -18,29 +18,258 @@
package org.apache.spark.security.aws;
import java.net.URI;
+import java.time.Duration;
+import java.time.Instant;
import java.util.Map;
import java.util.Set;
+import java.util.regex.Pattern;
+
+import com.google.common.annotations.VisibleForTesting;
+import software.amazon.awssdk.auth.credentials.AnonymousCredentialsProvider;
+import software.amazon.awssdk.core.exception.SdkException;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.sts.StsClient;
+import software.amazon.awssdk.services.sts.StsClientBuilder;
+import software.amazon.awssdk.services.sts.model.AssumeRoleWithWebIdentityRequest;
+import software.amazon.awssdk.services.sts.model.AssumeRoleWithWebIdentityResponse;
+import software.amazon.awssdk.services.sts.model.Credentials;
-import org.apache.spark.annotation.DeveloperApi;
import org.apache.spark.security.CredentialProvider;
import org.apache.spark.security.CredentialResolutionException;
import org.apache.spark.security.ServiceCredential;
import org.apache.spark.security.UserContext;
/**
- * :: DeveloperApi ::
- * AWS STS credential provider that exchanges an OIDC identity token for temporary
- * AWS credentials via {@code AssumeRoleWithWebIdentity}.
+ * A {@link CredentialProvider} that exchanges an OIDC identity token for temporary
+ * AWS credentials via the STS {@code AssumeRoleWithWebIdentity} API.
+ * <p>
+ * The returned {@link ServiceCredential} contains S3A-compatible Hadoop configuration
+ * properties ({@code fs.s3a.access.key}, {@code fs.s3a.secret.key},
+ * {@code fs.s3a.session.token}) that can be propagated to executors for accessing
+ * S3-compatible storage.
+ * <p>
+ * <b>Configuration keys</b> (passed via {@code spark.security.oidc.*}):
+ * <ul>
+ * <li>{@code spark.security.oidc.aws.roleArn} (required) -- the ARN of the IAM role
+ * to assume</li>
+ * <li>{@code spark.security.oidc.aws.sessionName} (optional) -- the role session name;
+ * defaults to a value derived from the user's principal or "spark-oidc"</li>
+ * <li>{@code spark.security.oidc.aws.durationSeconds} (optional) -- credential duration
+ * in seconds (900-43200); if unset, STS uses the role's default maximum</li>
+ * <li>{@code spark.security.oidc.aws.region} (optional) -- the AWS region for the STS
+ * endpoint; defaults to us-east-1 when only {@code stsEndpoint} is set. When neither
+ * {@code region} nor {@code stsEndpoint} is configured, the STS client falls back to
+ * the AWS SDK default region resolution (AWS_REGION / AWS_DEFAULT_REGION environment
+ * variables, then the ~/.aws/config profile).</li>
+ * <li>{@code spark.security.oidc.aws.stsEndpoint} (optional) -- a custom STS endpoint URL
+ * for non-AWS environments (MinIO, Ceph, LocalStack, etc.)</li>
+ * </ul>
+ * <p>
+ * <b>Security note:</b> The OIDC raw token is never included in log messages or
+ * exception messages. It is passed directly to the STS API call and discarded.
*
* @since 4.3.0
*/
-@DeveloperApi
public class AwsStsCredentialProvider implements CredentialProvider {
+ // Configuration key constants
+ static final String CONF_ROLE_ARN = "spark.security.oidc.aws.roleArn";
+ static final String CONF_SESSION_NAME = "spark.security.oidc.aws.sessionName";
+ static final String CONF_DURATION_SECONDS = "spark.security.oidc.aws.durationSeconds";
+ static final String CONF_REGION = "spark.security.oidc.aws.region";
+ static final String CONF_STS_ENDPOINT = "spark.security.oidc.aws.stsEndpoint";
+
+ /** Minimum duration allowed by STS AssumeRoleWithWebIdentity (15 minutes). */
+ static final int MIN_DURATION_SECONDS = 900;
+ /** Maximum duration allowed by STS AssumeRoleWithWebIdentity (12 hours). */
+ static final int MAX_DURATION_SECONDS = 43200;
+
+ private static final String DEFAULT_REGION = "us-east-1";
+ private static final String DEFAULT_SESSION_NAME = "spark-oidc";
+
+ /**
+ * Precompiled pattern matching characters that are NOT valid in STS session names.
+ * Valid characters are: alphanumeric, underscore, plus, equals, comma, period, at, hyphen.
+ */
+ private static final Pattern SESSION_NAME_INVALID_CHARS =
+ Pattern.compile("[^a-zA-Z0-9_+=,.@\\-]");
+
+ /**
+ * Immutable configuration holder that is safely published via the volatile
+ * {@link #config} field. All fields are set once during construction and are
+ * final, ensuring correct visibility across threads after init() completes.
+ */
+ static final class ResolvedConfig {
+ final String roleArn;
+ final String roleSessionName;
+ final Integer durationSeconds;
+ final Region resolvedRegion;
+ final URI endpointOverride;
+ final StsClient stsClient;
+
+ ResolvedConfig(String roleArn, String roleSessionName, Integer durationSeconds,
+ Region resolvedRegion, URI endpointOverride, StsClient stsClient) {
+ this.roleArn = roleArn;
+ this.roleSessionName = roleSessionName;
+ this.durationSeconds = durationSeconds;
+ this.resolvedRegion = resolvedRegion;
+ this.endpointOverride = endpointOverride;
+ this.stsClient = stsClient;
+ }
+ }
+
+ /** Safely published via volatile write in init(); read in resolve()/suggestedTtl(). */
+ private volatile ResolvedConfig config;
+
+ /**
+ * Default no-arg constructor used by {@link java.util.ServiceLoader}.
+ */
+ public AwsStsCredentialProvider() {
+ // ServiceLoader requires a public no-arg constructor
+ }
+
+ /**
+ * Package-private constructor for testing with an injected STS client.
+ * <p>
+ * This constructor is visible for testing only; production code must use the
+ * no-arg constructor followed by {@link #init(Map)}.
+ *
+ * @param stsClient the STS client to use (must not be null)
+ * @param roleArn the IAM role ARN (must not be null or blank)
+ * @param roleSessionName the session name (may be null for default)
+ * @param durationSeconds the credential duration in seconds (may be null)
+ */
+ @VisibleForTesting
+ AwsStsCredentialProvider(StsClient stsClient, String roleArn, String roleSessionName,
+ Integer durationSeconds) {
+ this.config = new ResolvedConfig(roleArn, roleSessionName, durationSeconds,
+ null, null, stsClient);
+ }
+
@Override
public void init(Map<String, String> conf) {
- throw new UnsupportedOperationException(
- "AwsStsCredentialProvider is a stub. Full implementation in SPARK-57898.");
+ if (this.config != null) {
+ throw new IllegalStateException("AwsStsCredentialProvider is already initialized");
+ }
+
+ String roleArn = conf.get(CONF_ROLE_ARN);
+ if (roleArn != null) {
+ roleArn = roleArn.trim();
+ }
+ if (roleArn == null || roleArn.isBlank()) {
+ throw new IllegalArgumentException(
+ "Configuration key '" + CONF_ROLE_ARN + "' is required but was not set. "
+ + "Specify the ARN of the IAM role to assume via AssumeRoleWithWebIdentity.");
+ }
+
+ String roleSessionName = conf.get(CONF_SESSION_NAME);
Review Comment:
Should we trim `roleSessionName` like as `roleArn`, `regionStr` and `stsEndpoint`?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57898][CORE] Implement AwsStsCredentialProvider (OIDC to STS credential exchange) [spark]
Posted by "yadavay-amzn (via GitHub)" <gi...@apache.org>.
yadavay-amzn commented on PR #57655:
URL: https://github.com/apache/spark/pull/57655#issuecomment-5260299348
@sarutak No problem, and thanks again to you and other reviewers for keeping the bar high, it has been a good learning experience for me.
Replaced the annotation with a plain comment.
For the AI disclosure, "No" was correct for the initial implementation, but, some of the commits addressing the review comments used AI assistance, I've updated the PR description to reflect that.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57898][CORE] Implement AwsStsCredentialProvider (OIDC to STS credential exchange) [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #57655:
URL: https://github.com/apache/spark/pull/57655#issuecomment-5247608156
For optional items (from my comments), yes, @yadavay-amzn .
BTW, `VisibleForTesting` tag might cause a Maven build failure. I'd recommend to remove Guava dependency. If needed, you need to provide `guava` provided dependency.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57898][CORE] Implement AwsStsCredentialProvider (OIDC to STS credential exchange) [spark]
Posted by "sarutak (via GitHub)" <gi...@apache.org>.
sarutak commented on code in PR #57655:
URL: https://github.com/apache/spark/pull/57655#discussion_r3742234539
##########
connector/credential-aws/src/main/java/org/apache/spark/security/aws/AwsStsCredentialProvider.java:
##########
@@ -18,29 +18,277 @@
package org.apache.spark.security.aws;
import java.net.URI;
+import java.time.Duration;
+import java.time.Instant;
import java.util.Map;
import java.util.Set;
+import java.util.regex.Pattern;
+import java.util.stream.Stream;
+
+import com.google.common.annotations.VisibleForTesting;
+import software.amazon.awssdk.auth.credentials.AnonymousCredentialsProvider;
+import software.amazon.awssdk.core.exception.SdkException;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.sts.StsClient;
+import software.amazon.awssdk.services.sts.StsClientBuilder;
+import software.amazon.awssdk.services.sts.model.AssumeRoleWithWebIdentityRequest;
+import software.amazon.awssdk.services.sts.model.AssumeRoleWithWebIdentityResponse;
+import software.amazon.awssdk.services.sts.model.Credentials;
-import org.apache.spark.annotation.DeveloperApi;
import org.apache.spark.security.CredentialProvider;
import org.apache.spark.security.CredentialResolutionException;
import org.apache.spark.security.ServiceCredential;
import org.apache.spark.security.UserContext;
/**
- * :: DeveloperApi ::
- * AWS STS credential provider that exchanges an OIDC identity token for temporary
- * AWS credentials via {@code AssumeRoleWithWebIdentity}.
+ * A {@link CredentialProvider} that exchanges an OIDC identity token for temporary
+ * AWS credentials via the STS {@code AssumeRoleWithWebIdentity} API.
+ * <p>
+ * The returned {@link ServiceCredential} contains S3A-compatible Hadoop configuration
+ * properties ({@code fs.s3a.access.key}, {@code fs.s3a.secret.key},
+ * {@code fs.s3a.session.token}) that can be propagated to executors for accessing
+ * S3-compatible storage.
+ * <p>
+ * <b>Configuration keys</b> (passed via {@code spark.security.oidc.*}):
+ * <ul>
+ * <li>{@code spark.security.oidc.aws.roleArn} (required) -- the ARN of the IAM role
+ * to assume</li>
+ * <li>{@code spark.security.oidc.aws.sessionName} (optional) -- the role session name;
+ * defaults to a value derived from the user's principal or "spark-oidc"</li>
+ * <li>{@code spark.security.oidc.aws.durationSeconds} (optional) -- credential duration
+ * in seconds (900-43200); if unset, STS uses the role's default maximum</li>
+ * <li>{@code spark.security.oidc.aws.region} (optional) -- the AWS region for the STS
+ * endpoint; defaults to us-east-1 when only {@code stsEndpoint} is set. When neither
+ * {@code region} nor {@code stsEndpoint} is configured, the STS client falls back to
+ * the AWS SDK default region resolution (AWS_REGION / AWS_DEFAULT_REGION environment
+ * variables, then the ~/.aws/config profile).</li>
+ * <li>{@code spark.security.oidc.aws.stsEndpoint} (optional) -- a custom STS endpoint URL
+ * for non-AWS environments (MinIO, Ceph, LocalStack, etc.)</li>
+ * </ul>
+ * <p>
+ * <b>Security note:</b> The OIDC raw token is never included in log messages or
+ * exception messages. It is passed directly to the STS API call and discarded.
*
* @since 4.3.0
*/
-@DeveloperApi
public class AwsStsCredentialProvider implements CredentialProvider {
+ // Configuration key constants
+ static final String CONF_ROLE_ARN = "spark.security.oidc.aws.roleArn";
+ static final String CONF_SESSION_NAME = "spark.security.oidc.aws.sessionName";
+ static final String CONF_DURATION_SECONDS = "spark.security.oidc.aws.durationSeconds";
+ static final String CONF_REGION = "spark.security.oidc.aws.region";
+ static final String CONF_STS_ENDPOINT = "spark.security.oidc.aws.stsEndpoint";
+
+ /** Minimum duration allowed by STS AssumeRoleWithWebIdentity (15 minutes). */
+ static final int MIN_DURATION_SECONDS = 900;
+ /** Maximum duration allowed by STS AssumeRoleWithWebIdentity (12 hours). */
+ static final int MAX_DURATION_SECONDS = 43200;
+
+ private static final String DEFAULT_REGION = "us-east-1";
+ private static final String DEFAULT_SESSION_NAME = "spark-oidc";
+
+ /**
+ * Precompiled pattern matching characters that are NOT valid in STS session names.
+ * Valid characters are: alphanumeric, underscore, plus, equals, comma, period, at, hyphen.
+ */
+ private static final Pattern SESSION_NAME_INVALID_CHARS =
+ Pattern.compile("[^a-zA-Z0-9_+=,.@\\-]");
+
+ /**
+ * Precompiled pattern for validating a configured STS session name.
+ * Must match {@code [\w+=,.@-]{2,64}} per AWS STS documentation.
+ */
+ private static final Pattern SESSION_NAME_VALID_PATTERN =
+ Pattern.compile("[\\w+=,.@\\-]{2,64}");
Review Comment:
`SESSION_NAME_VALID_PATTERN` uses `[\\w+=,.@\\-]{2,64}`, but Java's `\w` matches Unicode word characters (e.g., accented letters, CJK characters), not just ASCII. STS only accepts ASCII `[a-zA-Z0-9_+=,.@-]`.
This is inconsistent with `SESSION_NAME_INVALID_CHARS` which correctly uses the explicit ASCII character class `[^a-zA-Z0-9_+=,.@\\-]`.
A user setting `spark.security.oidc.aws.sessionName=café` would pass the validation in `init()` but be rejected by STS at resolve time -- defeating the fail-fast intent.
Could you change the validation pattern to match the sanitization pattern?
```java
private static final Pattern SESSION_NAME_VALID_PATTERN =
Pattern.compile("[a-zA-Z0-9_+=,.@\\-]{2,64}");
```
Also please update the related references:
- The error message at L182: `"must match [\\w+=,.@-]{2,64}"` -> `"must match [a-zA-Z0-9_+=,.@-]{2,64}"`
- The Javadoc for `SESSION_NAME_VALID_PATTERN`
- The Javadoc for `sanitizeSessionName`, which ironically says "We use an explicit character class rather than `\w`" while the validation pattern does the opposite
- The PR description mentions `[\w+=,.@-]{2,64}`. Please update that as well
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57898][CORE] Implement AwsStsCredentialProvider (OIDC to STS credential exchange) [spark]
Posted by "sarutak (via GitHub)" <gi...@apache.org>.
sarutak commented on PR #57655:
URL: https://github.com/apache/spark/pull/57655#issuecomment-5266238802
Merged. Thank you all!
I'll follow up with a small PR to extract common test helper methods and clean up the `// Visible for testing only.` line comments on the test constructor and `resolvedConfig()`. The Javadoc already states the intent, so the inline comments are redundant and mix two comment styles unnecessarily.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57898][CORE] Implement AwsStsCredentialProvider (OIDC to STS credential exchange) [spark]
Posted by "sarutak (via GitHub)" <gi...@apache.org>.
sarutak closed pull request #57655: [SPARK-57898][CORE] Implement AwsStsCredentialProvider (OIDC to STS credential exchange)
URL: https://github.com/apache/spark/pull/57655
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57898][CORE] Implement AwsStsCredentialProvider (OIDC to STS credential exchange) [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #57655:
URL: https://github.com/apache/spark/pull/57655#issuecomment-5170181686
No new blocking issues beyond what has already been raised. A few non-blocking suggestions to consider while reworking the PR after the rebase:
**1. Validate the `durationSeconds` range in `init()`**
`init()` only validates that `durationSeconds` is positive, but the valid range for STS `AssumeRoleWithWebIdentity` is **900–43200 seconds**. A value like `600` passes `init()` and then fails at the first `resolve()` call with an STS `ValidationError`. Validating the range in `init()` (fail-fast) would surface the misconfiguration much more clearly:
```java
if (durationSeconds < 900 || durationSeconds > 43200) {
throw new IllegalArgumentException(
"Configuration key '" + CONF_DURATION_SECONDS
+ "' must be between 900 and 43200 seconds, got: " + durationStr);
}
```
**2. `resolve()` neither checks nor uses `target`**
The SPI contract (`CredentialProvider#resolve`) states that `target` "must not be null", but the implementation validates only `user`. Not using `target` is reasonable for a role-based STS exchange, but for consistency with the `user` validation, consider either adding a null check or noting in the javadoc why `target` is intentionally unused.
**3. Token-redaction guarantee is only tested against mock messages**
The `SdkException` handler includes `e.getMessage()` in the wrapped message. The "token never appears in exception messages" assertion is only verified against a mock-generated message (`"Access denied for role"`), so in practice the guarantee relies on AWS never echoing the token in its error messages (which it normally does not — low actual risk). If you want to be defensive, you could filter the token string out of the message before wrapping.
**4. Test coverage gaps**
- `sanitizeSessionName` edge cases are untested: truncation of a principal longer than 64 chars, replacement of invalid characters with `-`, and the fallback to `DEFAULT_SESSION_NAME` when the sanitized result is shorter than 2 chars. Only the happy path (`alice@corp.example.com`) is covered.
- `init()` with `0` or a negative `durationSeconds` should assert `IllegalArgumentException` (currently only `"not-a-number"` is tested).
- `testServiceLoaderDiscovery` duplicates the same test already present in the stub suite on `master`; please dedupe after rebasing.
- Once `close()` is implemented (per the earlier comment), please add a test covering resource cleanup and the tolerated `resolve()` failure after `close()`.
On the plus side: the single-volatile-write publication of the immutable `ResolvedConfig` matches the SPI's concurrency requirements nicely, the `ArgumentCaptor`-based assertions on the STS request fields are thorough, and `@since 4.3.0` is consistent with the existing SPI and branch versioning policy.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57898][CORE] Implement AwsStsCredentialProvider (OIDC to STS credential exchange) [spark]
Posted by "yadavay-amzn (via GitHub)" <gi...@apache.org>.
yadavay-amzn commented on PR #57655:
URL: https://github.com/apache/spark/pull/57655#issuecomment-5222673398
Thanks again for reviewing @dongjoon-hyun
1. Cause-chain redaction: walks the full chain now, and drops the original entirely when the token appears anywhere; new test constructs a token planted only in a nested cause and asserts every layer is scrubbed.
2. `@DeveloperApi` removal is intentional (SPI implementation, not itself public API); called out in the PR description now, `@since 4.3.0` kept in the Javadoc.
3. Configured `sessionName` is validated against `[\w+=,.@-]{2,64}` in `init()`; IAE names the config key on any mismatch.
4. `@AfterEach` closes a suite-level provider field, so the tests going through real `init()` no longer leak `StsClient` handles.
5. `resolve()` catches `IllegalStateException` narrowly (post-close case) and wraps in `CredentialResolutionException`.
6. Docs follow-up is tracked as [SPARK-57901](https://issues.apache.org/jira/browse/SPARK-57901) (Add documentation about OIDC Credential Propagation); pointer added to the PR description. These 5 `spark.security.oidc.aws.*` keys will land there alongside the other OIDC keys.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57898][CORE] Implement AwsStsCredentialProvider (OIDC to STS credential exchange) [spark]
Posted by "sarutak (via GitHub)" <gi...@apache.org>.
sarutak commented on PR #57655:
URL: https://github.com/apache/spark/pull/57655#issuecomment-5163774168
Hi @yadavay-amzn, #57677 and #57679 have been merged. Could you rebase this PR to `master` and continue this PR work?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57898][CORE] Implement AwsStsCredentialProvider (OIDC to STS credential exchange) [spark]
Posted by "yadavay-amzn (via GitHub)" <gi...@apache.org>.
yadavay-amzn commented on PR #57655:
URL: https://github.com/apache/spark/pull/57655#issuecomment-5258690794
Thanks @dongjoon-hyun !
For the `VisibleForTesting` tag, this was requested by @sarutak in a previous round of review, so I want to double check before deciding how to proceed.
We can either drop the tag/annotation and leave a plain comment (like `// visible for testing`) to avoid the dependency addition or keep the tag and add the guava dependency.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57898][CORE] Implement AwsStsCredentialProvider (OIDC to STS credential exchange) [spark]
Posted by "dongjoon-hyun (via GitHub)" <gi...@apache.org>.
dongjoon-hyun commented on PR #57655:
URL: https://github.com/apache/spark/pull/57655#issuecomment-5220152797
Thank you for the contribution. Overall this looks well done — the single-volatile immutable config holder correctly satisfies the SPI thread-safety contract, the anonymous STS client is the right choice for `AssumeRoleWithWebIdentity`, and the test coverage is thorough. A few comments before merging:
1. **Token redaction does not cover the deep cause chain.** In the `SdkException` catch block, only the top-level `e.getMessage()` is checked and redacted; when redaction happens, `e.getCause()` is preserved as-is. If a nested cause message ever contains the token (e.g., an HTTP-level exception echoing the request body), it would still leak. Since this PR explicitly claims the token is never exposed, consider walking the cause chain, or dropping the cause entirely when the token is detected in the top-level message.
2. **`@DeveloperApi` removal.** The stub carried `@DeveloperApi`, which this PR removes while keeping `@since 4.3.0` in the javadoc. Removing it seems reasonable for a ServiceLoader-discovered implementation class, but could you confirm this is intentional and mention it in the PR description?
3. **Configured `sessionName` is not validated.** A principal-derived session name is sanitized against the STS constraint (`[\w+=,.@-]{2,64}`), but a user-configured `spark.security.oidc.aws.sessionName` is only trimmed and passed through. An invalid value would surface as an opaque STS error at resolve time; validating it in `init()` (fail-fast) would be friendlier.
4. **Resource leak in tests.** The tests that go through the real `init()` path (e.g., `testInitWithMinimumValidDurationSucceeds`, `testInitWithEndpointAndRegionResolvesCorrectly`, etc.) build real SDK HTTP clients and never close them. Please call `provider.close()` at the end of those tests.
5. **(Minor) `resolve()` after `close()` escapes as an unchecked exception.** A closed SDK client throws `IllegalStateException`, not `SdkException`, so it bypasses the catch and propagates unwrapped. The SPI contract allows in-flight `resolve()` to fail after `close()`, so this is not a violation — but catching `RuntimeException` and wrapping it into `CredentialResolutionException` would be more consistent for callers.
6. **(Minor) Docs.** This adds five user-facing `spark.security.oidc.aws.*` keys with no `docs/` update. If documentation is planned as a follow-up task of the SPIP, a pointer in the PR description would help.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57898][CORE] Implement AwsStsCredentialProvider (OIDC to STS credential exchange) [spark]
Posted by "sarutak (via GitHub)" <gi...@apache.org>.
sarutak commented on code in PR #57655:
URL: https://github.com/apache/spark/pull/57655#discussion_r3687938661
##########
connector/credential-aws/src/main/java/org/apache/spark/security/aws/AwsStsCredentialProvider.java:
##########
@@ -0,0 +1,316 @@
+/*
+ * 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.aws;
+
+import java.net.URI;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Map;
+import java.util.Set;
+
+import software.amazon.awssdk.auth.credentials.AnonymousCredentialsProvider;
+import software.amazon.awssdk.core.exception.SdkException;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.sts.StsClient;
+import software.amazon.awssdk.services.sts.StsClientBuilder;
+import software.amazon.awssdk.services.sts.model.AssumeRoleWithWebIdentityRequest;
+import software.amazon.awssdk.services.sts.model.AssumeRoleWithWebIdentityResponse;
+import software.amazon.awssdk.services.sts.model.Credentials;
+
+import org.apache.spark.annotation.DeveloperApi;
+import org.apache.spark.security.CredentialProvider;
+import org.apache.spark.security.CredentialResolutionException;
+import org.apache.spark.security.ServiceCredential;
+import org.apache.spark.security.UserContext;
+
+/**
+ * :: DeveloperApi ::
+ * A {@link CredentialProvider} that exchanges an OIDC identity token for temporary
+ * AWS credentials via the STS {@code AssumeRoleWithWebIdentity} API.
+ * <p>
+ * The returned {@link ServiceCredential} contains S3A-compatible Hadoop configuration
+ * properties ({@code fs.s3a.access.key}, {@code fs.s3a.secret.key},
+ * {@code fs.s3a.session.token}) that can be propagated to executors for accessing
+ * S3-compatible storage.
+ * <p>
+ * <b>Configuration keys</b> (passed via {@code spark.security.oidc.*}):
+ * <ul>
+ * <li>{@code spark.security.oidc.roleArn} (required) — the ARN of the IAM role to assume</li>
+ * <li>{@code spark.security.oidc.sessionName} (optional) — the role session name;
+ * defaults to a value derived from the user's principal or "spark-oidc"</li>
+ * <li>{@code spark.security.oidc.durationSeconds} (optional) — credential duration in seconds;
+ * if unset, STS uses the role's default maximum</li>
+ * <li>{@code spark.security.oidc.region} (optional) — the AWS region for the STS endpoint;
+ * defaults to us-east-1 when only {@code stsEndpoint} is set</li>
+ * <li>{@code spark.security.oidc.stsEndpoint} (optional) — a custom STS endpoint URL for
+ * non-AWS environments (MinIO, Ceph, LocalStack, etc.)</li>
+ * </ul>
+ * <p>
+ * <b>Security note:</b> The OIDC raw token is never included in log messages or
+ * exception messages. It is passed directly to the STS API call and discarded.
+ *
+ * @since 4.3.0
+ */
+@DeveloperApi
+public class AwsStsCredentialProvider implements CredentialProvider {
+
+ // Configuration key constants
+ static final String CONF_ROLE_ARN = "spark.security.oidc.roleArn";
Review Comment:
Could you rename to `spark.security.oidc.aws.*` to align with the SPIP doc?
##########
pom.xml:
##########
@@ -3519,6 +3519,13 @@
</modules>
</profile>
+ <profile>
+ <id>credential-aws</id>
+ <modules>
+ <module>connector/credential-aws</module>
Review Comment:
Since introducing a new Maven profile requires updates to many CI/release/lint scripts, I'd like to split the module scaffolding (SPARK-57897) into a separate PR as originally anticipated. I'll handle it. It will include:
- Module definition (`connector/credential-aws/pom.xml` with a stub provider)
- `pom.xml` profile, `SparkBuild.scala`, `modules.py`, `utils.py`
- `META-INF/services` registration
- CI/release/lint script updates:
- `.github/workflows/build_and_test.yml`
- `.github/workflows/maven_test.yml`
- `.github/workflows/python_hosted_runner_test.yml`
- `.github/workflows/benchmark.yml`
- `dev/create-release/release-build.sh`
- `dev/test-dependencies.sh`
- `dev/mima`
- `dev/lint-java`
- `dev/sbt-checkstyle`
- `dev/scalastyle`
- `docs/_plugins/build_api_docs.rb`
- `dev/spark-test-image-util/docs/build-docs`
Once that's merged, this PR can focus purely on the `AwsStsCredentialProvider` implementation and tests, removing the build wiring changes from its diff.
##########
connector/credential-aws/src/main/java/org/apache/spark/security/aws/AwsStsCredentialProvider.java:
##########
@@ -0,0 +1,316 @@
+/*
+ * 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.aws;
+
+import java.net.URI;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Map;
+import java.util.Set;
+
+import software.amazon.awssdk.auth.credentials.AnonymousCredentialsProvider;
+import software.amazon.awssdk.core.exception.SdkException;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.sts.StsClient;
+import software.amazon.awssdk.services.sts.StsClientBuilder;
+import software.amazon.awssdk.services.sts.model.AssumeRoleWithWebIdentityRequest;
+import software.amazon.awssdk.services.sts.model.AssumeRoleWithWebIdentityResponse;
+import software.amazon.awssdk.services.sts.model.Credentials;
+
+import org.apache.spark.annotation.DeveloperApi;
+import org.apache.spark.security.CredentialProvider;
+import org.apache.spark.security.CredentialResolutionException;
+import org.apache.spark.security.ServiceCredential;
+import org.apache.spark.security.UserContext;
+
+/**
+ * :: DeveloperApi ::
+ * A {@link CredentialProvider} that exchanges an OIDC identity token for temporary
+ * AWS credentials via the STS {@code AssumeRoleWithWebIdentity} API.
+ * <p>
+ * The returned {@link ServiceCredential} contains S3A-compatible Hadoop configuration
+ * properties ({@code fs.s3a.access.key}, {@code fs.s3a.secret.key},
+ * {@code fs.s3a.session.token}) that can be propagated to executors for accessing
+ * S3-compatible storage.
+ * <p>
+ * <b>Configuration keys</b> (passed via {@code spark.security.oidc.*}):
+ * <ul>
+ * <li>{@code spark.security.oidc.roleArn} (required) — the ARN of the IAM role to assume</li>
+ * <li>{@code spark.security.oidc.sessionName} (optional) — the role session name;
+ * defaults to a value derived from the user's principal or "spark-oidc"</li>
+ * <li>{@code spark.security.oidc.durationSeconds} (optional) — credential duration in seconds;
+ * if unset, STS uses the role's default maximum</li>
+ * <li>{@code spark.security.oidc.region} (optional) — the AWS region for the STS endpoint;
+ * defaults to us-east-1 when only {@code stsEndpoint} is set</li>
+ * <li>{@code spark.security.oidc.stsEndpoint} (optional) — a custom STS endpoint URL for
+ * non-AWS environments (MinIO, Ceph, LocalStack, etc.)</li>
+ * </ul>
+ * <p>
+ * <b>Security note:</b> The OIDC raw token is never included in log messages or
+ * exception messages. It is passed directly to the STS API call and discarded.
+ *
+ * @since 4.3.0
+ */
+@DeveloperApi
+public class AwsStsCredentialProvider implements CredentialProvider {
Review Comment:
`StsClient` implements `SdkAutoCloseable` and holds HTTP connection pools internally. The current implementation creates the client in `init()` but never closes it.
`CredentialProvider` uses an `init()` pattern where implementations construct long-lived resources. This means the SPI should provide a lifecycle hook for cleanup.
I'll open a PR to add `AutoCloseable` to `CredentialProvider` (with a default no-op `close()`) and wire `UserCredentialManager.stop()` to call `provider.close()`. Once that's merged, could you implement `close()` here to shut down the `StsClient`?
In the meantime, could you also add a re-initialization guard at the top of `init()`?
```java
if (this.config != null) {
throw new IllegalStateException("AwsStsCredentialProvider is already initialized");
}
```
##########
connector/credential-aws/src/main/java/org/apache/spark/security/aws/AwsStsCredentialProvider.java:
##########
@@ -0,0 +1,316 @@
+/*
+ * 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.aws;
+
+import java.net.URI;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Map;
+import java.util.Set;
+
+import software.amazon.awssdk.auth.credentials.AnonymousCredentialsProvider;
+import software.amazon.awssdk.core.exception.SdkException;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.sts.StsClient;
+import software.amazon.awssdk.services.sts.StsClientBuilder;
+import software.amazon.awssdk.services.sts.model.AssumeRoleWithWebIdentityRequest;
+import software.amazon.awssdk.services.sts.model.AssumeRoleWithWebIdentityResponse;
+import software.amazon.awssdk.services.sts.model.Credentials;
+
+import org.apache.spark.annotation.DeveloperApi;
+import org.apache.spark.security.CredentialProvider;
+import org.apache.spark.security.CredentialResolutionException;
+import org.apache.spark.security.ServiceCredential;
+import org.apache.spark.security.UserContext;
+
+/**
+ * :: DeveloperApi ::
+ * A {@link CredentialProvider} that exchanges an OIDC identity token for temporary
+ * AWS credentials via the STS {@code AssumeRoleWithWebIdentity} API.
+ * <p>
+ * The returned {@link ServiceCredential} contains S3A-compatible Hadoop configuration
+ * properties ({@code fs.s3a.access.key}, {@code fs.s3a.secret.key},
+ * {@code fs.s3a.session.token}) that can be propagated to executors for accessing
+ * S3-compatible storage.
+ * <p>
+ * <b>Configuration keys</b> (passed via {@code spark.security.oidc.*}):
+ * <ul>
+ * <li>{@code spark.security.oidc.roleArn} (required) — the ARN of the IAM role to assume</li>
+ * <li>{@code spark.security.oidc.sessionName} (optional) — the role session name;
+ * defaults to a value derived from the user's principal or "spark-oidc"</li>
+ * <li>{@code spark.security.oidc.durationSeconds} (optional) — credential duration in seconds;
+ * if unset, STS uses the role's default maximum</li>
+ * <li>{@code spark.security.oidc.region} (optional) — the AWS region for the STS endpoint;
+ * defaults to us-east-1 when only {@code stsEndpoint} is set</li>
+ * <li>{@code spark.security.oidc.stsEndpoint} (optional) — a custom STS endpoint URL for
+ * non-AWS environments (MinIO, Ceph, LocalStack, etc.)</li>
+ * </ul>
+ * <p>
+ * <b>Security note:</b> The OIDC raw token is never included in log messages or
+ * exception messages. It is passed directly to the STS API call and discarded.
+ *
+ * @since 4.3.0
+ */
+@DeveloperApi
+public class AwsStsCredentialProvider implements CredentialProvider {
+
+ // Configuration key constants
+ static final String CONF_ROLE_ARN = "spark.security.oidc.roleArn";
+ static final String CONF_SESSION_NAME = "spark.security.oidc.sessionName";
+ static final String CONF_DURATION_SECONDS = "spark.security.oidc.durationSeconds";
+ static final String CONF_REGION = "spark.security.oidc.region";
+ static final String CONF_STS_ENDPOINT = "spark.security.oidc.stsEndpoint";
+
+ private static final String DEFAULT_REGION = "us-east-1";
+ private static final String DEFAULT_SESSION_NAME = "spark-oidc";
+
+ /**
+ * Immutable configuration holder that is safely published via the volatile
+ * {@link #config} field. All fields are set once during construction and are
+ * final, ensuring correct visibility across threads after init() completes.
+ */
+ static final class ResolvedConfig {
+ final String roleArn;
+ final String roleSessionName;
+ final Integer durationSeconds;
+ final Region resolvedRegion;
+ final URI endpointOverride;
+ final StsClient stsClient;
+
+ ResolvedConfig(String roleArn, String roleSessionName, Integer durationSeconds,
+ Region resolvedRegion, URI endpointOverride, StsClient stsClient) {
+ this.roleArn = roleArn;
+ this.roleSessionName = roleSessionName;
+ this.durationSeconds = durationSeconds;
+ this.resolvedRegion = resolvedRegion;
+ this.endpointOverride = endpointOverride;
+ this.stsClient = stsClient;
+ }
+ }
+
+ /** Safely published via volatile write in init(); read in resolve()/suggestedTtl(). */
+ private volatile ResolvedConfig config;
+
+ /**
+ * Default no-arg constructor used by {@link java.util.ServiceLoader}.
+ */
+ public AwsStsCredentialProvider() {
+ // ServiceLoader requires a public no-arg constructor
+ }
+
+ /**
+ * Package-private constructor for testing with an injected STS client.
+ *
+ * @param stsClient the STS client to use (must not be null)
+ * @param roleArn the IAM role ARN (must not be null or blank)
+ * @param roleSessionName the session name (may be null for default)
+ * @param durationSeconds the credential duration in seconds (may be null)
+ */
+ AwsStsCredentialProvider(StsClient stsClient, String roleArn, String roleSessionName,
Review Comment:
Please add `@VisibleForTesting` to this package-private constructor.
##########
connector/credential-aws/src/main/java/org/apache/spark/security/aws/AwsStsCredentialProvider.java:
##########
@@ -0,0 +1,316 @@
+/*
+ * 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.aws;
+
+import java.net.URI;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Map;
+import java.util.Set;
+
+import software.amazon.awssdk.auth.credentials.AnonymousCredentialsProvider;
+import software.amazon.awssdk.core.exception.SdkException;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.sts.StsClient;
+import software.amazon.awssdk.services.sts.StsClientBuilder;
+import software.amazon.awssdk.services.sts.model.AssumeRoleWithWebIdentityRequest;
+import software.amazon.awssdk.services.sts.model.AssumeRoleWithWebIdentityResponse;
+import software.amazon.awssdk.services.sts.model.Credentials;
+
+import org.apache.spark.annotation.DeveloperApi;
+import org.apache.spark.security.CredentialProvider;
+import org.apache.spark.security.CredentialResolutionException;
+import org.apache.spark.security.ServiceCredential;
+import org.apache.spark.security.UserContext;
+
+/**
+ * :: DeveloperApi ::
+ * A {@link CredentialProvider} that exchanges an OIDC identity token for temporary
+ * AWS credentials via the STS {@code AssumeRoleWithWebIdentity} API.
+ * <p>
+ * The returned {@link ServiceCredential} contains S3A-compatible Hadoop configuration
+ * properties ({@code fs.s3a.access.key}, {@code fs.s3a.secret.key},
+ * {@code fs.s3a.session.token}) that can be propagated to executors for accessing
+ * S3-compatible storage.
+ * <p>
+ * <b>Configuration keys</b> (passed via {@code spark.security.oidc.*}):
+ * <ul>
+ * <li>{@code spark.security.oidc.roleArn} (required) — the ARN of the IAM role to assume</li>
+ * <li>{@code spark.security.oidc.sessionName} (optional) — the role session name;
+ * defaults to a value derived from the user's principal or "spark-oidc"</li>
+ * <li>{@code spark.security.oidc.durationSeconds} (optional) — credential duration in seconds;
+ * if unset, STS uses the role's default maximum</li>
+ * <li>{@code spark.security.oidc.region} (optional) — the AWS region for the STS endpoint;
+ * defaults to us-east-1 when only {@code stsEndpoint} is set</li>
+ * <li>{@code spark.security.oidc.stsEndpoint} (optional) — a custom STS endpoint URL for
+ * non-AWS environments (MinIO, Ceph, LocalStack, etc.)</li>
+ * </ul>
+ * <p>
+ * <b>Security note:</b> The OIDC raw token is never included in log messages or
+ * exception messages. It is passed directly to the STS API call and discarded.
+ *
+ * @since 4.3.0
+ */
+@DeveloperApi
+public class AwsStsCredentialProvider implements CredentialProvider {
+
+ // Configuration key constants
+ static final String CONF_ROLE_ARN = "spark.security.oidc.roleArn";
+ static final String CONF_SESSION_NAME = "spark.security.oidc.sessionName";
+ static final String CONF_DURATION_SECONDS = "spark.security.oidc.durationSeconds";
+ static final String CONF_REGION = "spark.security.oidc.region";
+ static final String CONF_STS_ENDPOINT = "spark.security.oidc.stsEndpoint";
+
+ private static final String DEFAULT_REGION = "us-east-1";
+ private static final String DEFAULT_SESSION_NAME = "spark-oidc";
+
+ /**
+ * Immutable configuration holder that is safely published via the volatile
+ * {@link #config} field. All fields are set once during construction and are
+ * final, ensuring correct visibility across threads after init() completes.
+ */
+ static final class ResolvedConfig {
+ final String roleArn;
+ final String roleSessionName;
+ final Integer durationSeconds;
+ final Region resolvedRegion;
+ final URI endpointOverride;
+ final StsClient stsClient;
+
+ ResolvedConfig(String roleArn, String roleSessionName, Integer durationSeconds,
+ Region resolvedRegion, URI endpointOverride, StsClient stsClient) {
+ this.roleArn = roleArn;
+ this.roleSessionName = roleSessionName;
+ this.durationSeconds = durationSeconds;
+ this.resolvedRegion = resolvedRegion;
+ this.endpointOverride = endpointOverride;
+ this.stsClient = stsClient;
+ }
+ }
+
+ /** Safely published via volatile write in init(); read in resolve()/suggestedTtl(). */
+ private volatile ResolvedConfig config;
+
+ /**
+ * Default no-arg constructor used by {@link java.util.ServiceLoader}.
+ */
+ public AwsStsCredentialProvider() {
+ // ServiceLoader requires a public no-arg constructor
+ }
+
+ /**
+ * Package-private constructor for testing with an injected STS client.
+ *
+ * @param stsClient the STS client to use (must not be null)
+ * @param roleArn the IAM role ARN (must not be null or blank)
+ * @param roleSessionName the session name (may be null for default)
+ * @param durationSeconds the credential duration in seconds (may be null)
+ */
+ AwsStsCredentialProvider(StsClient stsClient, String roleArn, String roleSessionName,
+ Integer durationSeconds) {
+ this.config = new ResolvedConfig(roleArn, roleSessionName, durationSeconds,
+ null, null, stsClient);
+ }
+
+ @Override
+ public void init(Map<String, String> conf) {
+ String roleArn = conf.get(CONF_ROLE_ARN);
+ if (roleArn == null || roleArn.isBlank()) {
+ throw new IllegalArgumentException(
+ "Configuration key '" + CONF_ROLE_ARN + "' is required but was not set. "
+ + "Specify the ARN of the IAM role to assume via AssumeRoleWithWebIdentity.");
+ }
+
+ String roleSessionName = conf.get(CONF_SESSION_NAME);
+ String regionStr = conf.get(CONF_REGION);
+ String stsEndpoint = conf.get(CONF_STS_ENDPOINT);
+
+ Integer durationSeconds = null;
+ String durationStr = conf.get(CONF_DURATION_SECONDS);
+ if (durationStr != null && !durationStr.isBlank()) {
+ try {
+ durationSeconds = Integer.parseInt(durationStr.trim());
+ if (durationSeconds <= 0) {
+ throw new IllegalArgumentException(
+ "Configuration key '" + CONF_DURATION_SECONDS
+ + "' must be a positive integer, got: " + durationStr);
+ }
+ } catch (NumberFormatException e) {
+ throw new IllegalArgumentException(
+ "Configuration key '" + CONF_DURATION_SECONDS
+ + "' must be a valid integer, got: " + durationStr, e);
+ }
+ }
+
+ // Resolve the region and endpoint before building the client
+ Region resolvedRegion = resolveRegion(regionStr, stsEndpoint);
+ URI endpointOverride = resolveEndpoint(stsEndpoint);
+
+ StsClient stsClient = buildStsClient(resolvedRegion, endpointOverride);
+
+ // Single volatile write publishes all configuration atomically
+ this.config = new ResolvedConfig(roleArn, roleSessionName, durationSeconds,
+ resolvedRegion, endpointOverride, stsClient);
+ }
+
+ /**
+ * Resolves the AWS region based on explicit configuration and endpoint presence.
+ * When a custom endpoint is provided without an explicit region, defaults to us-east-1.
+ */
+ private static Region resolveRegion(String regionStr, String stsEndpoint) {
+ if (regionStr != null && !regionStr.isBlank()) {
+ return Region.of(regionStr);
+ } else if (stsEndpoint != null && !stsEndpoint.isBlank()) {
+ // When a custom endpoint is set but no explicit region, use a default region.
+ // The region is required by the SDK but not meaningful for non-AWS endpoints.
+ return Region.of(DEFAULT_REGION);
+ }
+ return null;
+ }
+
+ /**
+ * Resolves the endpoint override URI from configuration.
+ */
+ private static URI resolveEndpoint(String stsEndpoint) {
+ if (stsEndpoint != null && !stsEndpoint.isBlank()) {
+ return URI.create(stsEndpoint);
+ }
+ return null;
+ }
+
+ /**
+ * Builds the STS client with the resolved region and endpoint.
+ */
+ private static StsClient buildStsClient(Region resolvedRegion, URI endpointOverride) {
+ StsClientBuilder builder = StsClient.builder()
+ .credentialsProvider(AnonymousCredentialsProvider.create());
Review Comment:
A brief comment explaining why anonymous credentials are used would help future readers:
```java
// AssumeRoleWithWebIdentity does not require AWS credentials;
// the OIDC token itself serves as the authentication mechanism.
.credentialsProvider(AnonymousCredentialsProvider.create())
```
##########
connector/credential-aws/src/main/java/org/apache/spark/security/aws/AwsStsCredentialProvider.java:
##########
@@ -0,0 +1,316 @@
+/*
+ * 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.aws;
+
+import java.net.URI;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Map;
+import java.util.Set;
+
+import software.amazon.awssdk.auth.credentials.AnonymousCredentialsProvider;
+import software.amazon.awssdk.core.exception.SdkException;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.sts.StsClient;
+import software.amazon.awssdk.services.sts.StsClientBuilder;
+import software.amazon.awssdk.services.sts.model.AssumeRoleWithWebIdentityRequest;
+import software.amazon.awssdk.services.sts.model.AssumeRoleWithWebIdentityResponse;
+import software.amazon.awssdk.services.sts.model.Credentials;
+
+import org.apache.spark.annotation.DeveloperApi;
+import org.apache.spark.security.CredentialProvider;
+import org.apache.spark.security.CredentialResolutionException;
+import org.apache.spark.security.ServiceCredential;
+import org.apache.spark.security.UserContext;
+
+/**
+ * :: DeveloperApi ::
+ * A {@link CredentialProvider} that exchanges an OIDC identity token for temporary
+ * AWS credentials via the STS {@code AssumeRoleWithWebIdentity} API.
+ * <p>
+ * The returned {@link ServiceCredential} contains S3A-compatible Hadoop configuration
+ * properties ({@code fs.s3a.access.key}, {@code fs.s3a.secret.key},
+ * {@code fs.s3a.session.token}) that can be propagated to executors for accessing
+ * S3-compatible storage.
+ * <p>
+ * <b>Configuration keys</b> (passed via {@code spark.security.oidc.*}):
+ * <ul>
+ * <li>{@code spark.security.oidc.roleArn} (required) — the ARN of the IAM role to assume</li>
+ * <li>{@code spark.security.oidc.sessionName} (optional) — the role session name;
+ * defaults to a value derived from the user's principal or "spark-oidc"</li>
+ * <li>{@code spark.security.oidc.durationSeconds} (optional) — credential duration in seconds;
+ * if unset, STS uses the role's default maximum</li>
+ * <li>{@code spark.security.oidc.region} (optional) — the AWS region for the STS endpoint;
+ * defaults to us-east-1 when only {@code stsEndpoint} is set</li>
+ * <li>{@code spark.security.oidc.stsEndpoint} (optional) — a custom STS endpoint URL for
+ * non-AWS environments (MinIO, Ceph, LocalStack, etc.)</li>
+ * </ul>
+ * <p>
+ * <b>Security note:</b> The OIDC raw token is never included in log messages or
+ * exception messages. It is passed directly to the STS API call and discarded.
+ *
+ * @since 4.3.0
+ */
+@DeveloperApi
+public class AwsStsCredentialProvider implements CredentialProvider {
+
+ // Configuration key constants
+ static final String CONF_ROLE_ARN = "spark.security.oidc.roleArn";
+ static final String CONF_SESSION_NAME = "spark.security.oidc.sessionName";
+ static final String CONF_DURATION_SECONDS = "spark.security.oidc.durationSeconds";
+ static final String CONF_REGION = "spark.security.oidc.region";
+ static final String CONF_STS_ENDPOINT = "spark.security.oidc.stsEndpoint";
+
+ private static final String DEFAULT_REGION = "us-east-1";
+ private static final String DEFAULT_SESSION_NAME = "spark-oidc";
+
+ /**
+ * Immutable configuration holder that is safely published via the volatile
+ * {@link #config} field. All fields are set once during construction and are
+ * final, ensuring correct visibility across threads after init() completes.
+ */
+ static final class ResolvedConfig {
+ final String roleArn;
+ final String roleSessionName;
+ final Integer durationSeconds;
+ final Region resolvedRegion;
+ final URI endpointOverride;
+ final StsClient stsClient;
+
+ ResolvedConfig(String roleArn, String roleSessionName, Integer durationSeconds,
+ Region resolvedRegion, URI endpointOverride, StsClient stsClient) {
+ this.roleArn = roleArn;
+ this.roleSessionName = roleSessionName;
+ this.durationSeconds = durationSeconds;
+ this.resolvedRegion = resolvedRegion;
+ this.endpointOverride = endpointOverride;
+ this.stsClient = stsClient;
+ }
+ }
+
+ /** Safely published via volatile write in init(); read in resolve()/suggestedTtl(). */
+ private volatile ResolvedConfig config;
+
+ /**
+ * Default no-arg constructor used by {@link java.util.ServiceLoader}.
+ */
+ public AwsStsCredentialProvider() {
+ // ServiceLoader requires a public no-arg constructor
+ }
+
+ /**
+ * Package-private constructor for testing with an injected STS client.
+ *
+ * @param stsClient the STS client to use (must not be null)
+ * @param roleArn the IAM role ARN (must not be null or blank)
+ * @param roleSessionName the session name (may be null for default)
+ * @param durationSeconds the credential duration in seconds (may be null)
+ */
+ AwsStsCredentialProvider(StsClient stsClient, String roleArn, String roleSessionName,
+ Integer durationSeconds) {
+ this.config = new ResolvedConfig(roleArn, roleSessionName, durationSeconds,
+ null, null, stsClient);
+ }
+
+ @Override
+ public void init(Map<String, String> conf) {
+ String roleArn = conf.get(CONF_ROLE_ARN);
+ if (roleArn == null || roleArn.isBlank()) {
+ throw new IllegalArgumentException(
+ "Configuration key '" + CONF_ROLE_ARN + "' is required but was not set. "
+ + "Specify the ARN of the IAM role to assume via AssumeRoleWithWebIdentity.");
+ }
+
+ String roleSessionName = conf.get(CONF_SESSION_NAME);
+ String regionStr = conf.get(CONF_REGION);
+ String stsEndpoint = conf.get(CONF_STS_ENDPOINT);
+
+ Integer durationSeconds = null;
+ String durationStr = conf.get(CONF_DURATION_SECONDS);
+ if (durationStr != null && !durationStr.isBlank()) {
+ try {
+ durationSeconds = Integer.parseInt(durationStr.trim());
+ if (durationSeconds <= 0) {
+ throw new IllegalArgumentException(
+ "Configuration key '" + CONF_DURATION_SECONDS
+ + "' must be a positive integer, got: " + durationStr);
+ }
+ } catch (NumberFormatException e) {
+ throw new IllegalArgumentException(
+ "Configuration key '" + CONF_DURATION_SECONDS
+ + "' must be a valid integer, got: " + durationStr, e);
+ }
+ }
+
+ // Resolve the region and endpoint before building the client
+ Region resolvedRegion = resolveRegion(regionStr, stsEndpoint);
+ URI endpointOverride = resolveEndpoint(stsEndpoint);
+
+ StsClient stsClient = buildStsClient(resolvedRegion, endpointOverride);
+
+ // Single volatile write publishes all configuration atomically
+ this.config = new ResolvedConfig(roleArn, roleSessionName, durationSeconds,
+ resolvedRegion, endpointOverride, stsClient);
+ }
+
+ /**
+ * Resolves the AWS region based on explicit configuration and endpoint presence.
+ * When a custom endpoint is provided without an explicit region, defaults to us-east-1.
+ */
+ private static Region resolveRegion(String regionStr, String stsEndpoint) {
+ if (regionStr != null && !regionStr.isBlank()) {
+ return Region.of(regionStr);
+ } else if (stsEndpoint != null && !stsEndpoint.isBlank()) {
+ // When a custom endpoint is set but no explicit region, use a default region.
+ // The region is required by the SDK but not meaningful for non-AWS endpoints.
+ return Region.of(DEFAULT_REGION);
+ }
+ return null;
+ }
+
+ /**
+ * Resolves the endpoint override URI from configuration.
+ */
+ private static URI resolveEndpoint(String stsEndpoint) {
+ if (stsEndpoint != null && !stsEndpoint.isBlank()) {
+ return URI.create(stsEndpoint);
+ }
+ return null;
+ }
+
+ /**
+ * Builds the STS client with the resolved region and endpoint.
+ */
+ private static StsClient buildStsClient(Region resolvedRegion, URI endpointOverride) {
+ StsClientBuilder builder = StsClient.builder()
+ .credentialsProvider(AnonymousCredentialsProvider.create());
+
+ if (resolvedRegion != null) {
+ builder.region(resolvedRegion);
+ }
+
+ if (endpointOverride != null) {
+ builder.endpointOverride(endpointOverride);
+ }
+
+ return builder.build();
+ }
+
+ /**
+ * Returns the resolved configuration for testing purposes.
+ * Package-private visibility allows test assertions on resolved region/endpoint.
+ */
+ ResolvedConfig resolvedConfig() {
+ return config;
+ }
+
+ @Override
+ public Set<String> supportedSchemes() {
+ return Set.of("s3a");
+ }
+
+ @Override
+ public ServiceCredential resolve(UserContext user, URI target)
+ throws CredentialResolutionException {
+ if (user == null) {
+ throw new CredentialResolutionException(
+ "UserContext must not be null when resolving AWS credentials");
+ }
+ if (user.getRawToken() == null || user.getRawToken().isBlank()) {
+ throw new CredentialResolutionException(
+ "UserContext raw token must not be null or blank; cannot perform "
+ + "AssumeRoleWithWebIdentity without an identity token");
+ }
+
+ ResolvedConfig cfg = this.config;
+ if (cfg == null) {
+ throw new CredentialResolutionException("resolve() called before init()");
+ }
+ String sessionName = cfg.roleSessionName;
+ if (sessionName == null || sessionName.isBlank()) {
+ // Derive from principal, sanitizing for STS session name constraints
+ // (alphanumeric, =,.@- only, max 64 chars)
+ String principal = user.getPrincipal();
+ if (principal != null && !principal.isBlank()) {
+ sessionName = sanitizeSessionName(principal);
+ } else {
+ sessionName = DEFAULT_SESSION_NAME;
+ }
+ }
+
+ try {
+ AssumeRoleWithWebIdentityRequest.Builder reqBuilder =
+ AssumeRoleWithWebIdentityRequest.builder()
+ .roleArn(cfg.roleArn)
+ .roleSessionName(sessionName)
+ .webIdentityToken(user.getRawToken());
+
+ if (cfg.durationSeconds != null) {
+ reqBuilder.durationSeconds(cfg.durationSeconds);
+ }
+
+ AssumeRoleWithWebIdentityResponse response =
+ cfg.stsClient.assumeRoleWithWebIdentity(reqBuilder.build());
+
+ Credentials creds = response.credentials();
+ if (creds == null || creds.accessKeyId() == null
+ || creds.secretAccessKey() == null || creds.sessionToken() == null) {
+ throw new CredentialResolutionException(
+ "STS returned incomplete credentials for role '" + cfg.roleArn + "'");
+ }
+
+ Map<String, String> properties = Map.of(
+ "fs.s3a.access.key", creds.accessKeyId(),
+ "fs.s3a.secret.key", creds.secretAccessKey(),
+ "fs.s3a.session.token", creds.sessionToken()
+ );
+
+ Instant expiration = creds.expiration();
+ return new ServiceCredential(properties, expiration);
+ } catch (SdkException e) {
+ // SECURITY: Never include the token in exception messages
+ throw new CredentialResolutionException(
+ "Failed to assume role '" + cfg.roleArn + "' via AssumeRoleWithWebIdentity: "
+ + e.getMessage(), e);
+ }
+ }
+
+ @Override
+ public Duration suggestedTtl() {
+ ResolvedConfig cfg = this.config;
+ if (cfg != null && cfg.durationSeconds != null) {
+ return Duration.ofSeconds(cfg.durationSeconds);
+ }
+ return Duration.ofMinutes(15);
+ }
+
+ /**
+ * Sanitizes a principal string to be valid as an STS role session name.
+ * STS session names must match [\w+=,.@-]{2,64}.
+ */
+ private static String sanitizeSessionName(String principal) {
+ String sanitized = principal.replaceAll("[^\\w+=,.@\\-]", "-");
Review Comment:
`\\w` in Java regex is locale-independent by default (matches `[a-zA-Z0-9_]`), but using an explicit character class makes the intent clearer and removes any doubt:
```java
String sanitized = principal.replaceAll("[^a-zA-Z0-9_+=,.@\\-]", "-");
```
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57898][CORE] Implement AwsStsCredentialProvider (OIDC to STS credential exchange) [spark]
Posted by "sarutak (via GitHub)" <gi...@apache.org>.
sarutak commented on code in PR #57655:
URL: https://github.com/apache/spark/pull/57655#discussion_r3688004287
##########
connector/credential-aws/src/main/java/org/apache/spark/security/aws/AwsStsCredentialProvider.java:
##########
@@ -0,0 +1,316 @@
+/*
+ * 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.aws;
+
+import java.net.URI;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Map;
+import java.util.Set;
+
+import software.amazon.awssdk.auth.credentials.AnonymousCredentialsProvider;
+import software.amazon.awssdk.core.exception.SdkException;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.sts.StsClient;
+import software.amazon.awssdk.services.sts.StsClientBuilder;
+import software.amazon.awssdk.services.sts.model.AssumeRoleWithWebIdentityRequest;
+import software.amazon.awssdk.services.sts.model.AssumeRoleWithWebIdentityResponse;
+import software.amazon.awssdk.services.sts.model.Credentials;
+
+import org.apache.spark.annotation.DeveloperApi;
+import org.apache.spark.security.CredentialProvider;
+import org.apache.spark.security.CredentialResolutionException;
+import org.apache.spark.security.ServiceCredential;
+import org.apache.spark.security.UserContext;
+
+/**
+ * :: DeveloperApi ::
+ * A {@link CredentialProvider} that exchanges an OIDC identity token for temporary
+ * AWS credentials via the STS {@code AssumeRoleWithWebIdentity} API.
+ * <p>
+ * The returned {@link ServiceCredential} contains S3A-compatible Hadoop configuration
+ * properties ({@code fs.s3a.access.key}, {@code fs.s3a.secret.key},
+ * {@code fs.s3a.session.token}) that can be propagated to executors for accessing
+ * S3-compatible storage.
+ * <p>
+ * <b>Configuration keys</b> (passed via {@code spark.security.oidc.*}):
+ * <ul>
+ * <li>{@code spark.security.oidc.roleArn} (required) — the ARN of the IAM role to assume</li>
+ * <li>{@code spark.security.oidc.sessionName} (optional) — the role session name;
+ * defaults to a value derived from the user's principal or "spark-oidc"</li>
+ * <li>{@code spark.security.oidc.durationSeconds} (optional) — credential duration in seconds;
+ * if unset, STS uses the role's default maximum</li>
+ * <li>{@code spark.security.oidc.region} (optional) — the AWS region for the STS endpoint;
+ * defaults to us-east-1 when only {@code stsEndpoint} is set</li>
+ * <li>{@code spark.security.oidc.stsEndpoint} (optional) — a custom STS endpoint URL for
+ * non-AWS environments (MinIO, Ceph, LocalStack, etc.)</li>
+ * </ul>
+ * <p>
+ * <b>Security note:</b> The OIDC raw token is never included in log messages or
+ * exception messages. It is passed directly to the STS API call and discarded.
+ *
+ * @since 4.3.0
+ */
+@DeveloperApi
+public class AwsStsCredentialProvider implements CredentialProvider {
Review Comment:
`StsClient` implements `SdkAutoCloseable` and holds HTTP connection pools internally. The current implementation creates the client in `init()` but never closes it.
`CredentialProvider` uses an `init()` pattern where implementations construct long-lived resources. This means the SPI should provide a lifecycle hook for cleanup.
I'll open a PR to add `AutoCloseable` to `CredentialProvider` (with a default no-op `close()`) and wire `UserCredentialManager.stop()` to call `provider.close()`. Once that's merged, could you implement `close()` here to shut down the `StsClient`?
In the meantime, could you also add a re-initialization guard at the top of `init()`?
```java
if (this.config != null) {
throw new IllegalStateException("AwsStsCredentialProvider is already initialized");
}
```
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57898][CORE] Implement AwsStsCredentialProvider (OIDC to STS credential exchange) [spark]
Posted by "sarutak (via GitHub)" <gi...@apache.org>.
sarutak commented on code in PR #57655:
URL: https://github.com/apache/spark/pull/57655#discussion_r3713182150
##########
connector/credential-aws/src/main/java/org/apache/spark/security/aws/AwsStsCredentialProvider.java:
##########
@@ -51,7 +254,102 @@ public Set<String> supportedSchemes() {
@Override
public ServiceCredential resolve(UserContext user, URI target)
throws CredentialResolutionException {
- throw new UnsupportedOperationException(
- "AwsStsCredentialProvider is a stub. Full implementation in SPARK-57898.");
+ if (user == null) {
+ throw new CredentialResolutionException(
+ "UserContext must not be null when resolving AWS credentials");
+ }
+ if (target == null) {
+ throw new CredentialResolutionException(
+ "Target URI must not be null when resolving AWS credentials");
+ }
+ if (user.getRawToken() == null || user.getRawToken().isBlank()) {
+ throw new CredentialResolutionException(
+ "UserContext raw token must not be null or blank; cannot perform "
+ + "AssumeRoleWithWebIdentity without an identity token");
+ }
+
+ ResolvedConfig cfg = this.config;
+ if (cfg == null) {
+ throw new CredentialResolutionException("resolve() called before init()");
+ }
+ String sessionName = cfg.roleSessionName;
+ if (sessionName == null || sessionName.isBlank()) {
+ // Derive from principal, sanitizing for STS session name constraints
+ // (alphanumeric, =,.@- only, max 64 chars)
+ String principal = user.getPrincipal();
+ if (principal != null && !principal.isBlank()) {
+ sessionName = sanitizeSessionName(principal);
+ } else {
+ sessionName = DEFAULT_SESSION_NAME;
+ }
+ }
+
+ String rawToken = user.getRawToken();
+ try {
+ AssumeRoleWithWebIdentityRequest.Builder reqBuilder =
+ AssumeRoleWithWebIdentityRequest.builder()
+ .roleArn(cfg.roleArn)
+ .roleSessionName(sessionName)
+ .webIdentityToken(rawToken);
+
+ if (cfg.durationSeconds != null) {
+ reqBuilder.durationSeconds(cfg.durationSeconds);
+ }
+
+ AssumeRoleWithWebIdentityResponse response =
+ cfg.stsClient.assumeRoleWithWebIdentity(reqBuilder.build());
+
+ Credentials creds = response.credentials();
+ if (creds == null || creds.accessKeyId() == null
+ || creds.secretAccessKey() == null || creds.sessionToken() == null) {
+ throw new CredentialResolutionException(
+ "STS returned incomplete credentials for role '" + cfg.roleArn + "'");
+ }
+
+ Map<String, String> properties = Map.of(
+ "fs.s3a.access.key", creds.accessKeyId(),
+ "fs.s3a.secret.key", creds.secretAccessKey(),
+ "fs.s3a.session.token", creds.sessionToken()
+ );
+
+ Instant expiration = creds.expiration();
+ return new ServiceCredential(properties, expiration);
+ } catch (SdkException e) {
+ // SECURITY: Never include the token in exception messages.
+ // Defensively strip any occurrence of the raw token from the STS error message
+ // in case the service accidentally echoed it.
+ String errorMsg = e.getMessage();
+ if (errorMsg != null && rawToken != null && errorMsg.contains(rawToken)) {
+ errorMsg = errorMsg.replace(rawToken, "[REDACTED]");
Review Comment:
The outer `CredentialResolutionException` message is correctly sanitized, but the original `SdkException` (which may contain the raw token in its message) is passed as the `cause`. Any logging framework that prints full stack traces, including `getCause().getMessage()`, will expose the token.
For example, if STS returns `"Invalid identity token: <TOKEN>"`, the current code produces:
```
CredentialResolutionException: Failed to assume role '...' via AssumeRoleWithWebIdentity: Invalid identity token: [REDACTED]
Caused by: StsException: Invalid identity token: eyJhbGciOi... <-- LEAKED
```
The existing test `testTokenRedactedFromStsErrorMessage` only asserts on `ex.getMessage()` but not on `ex.getCause().getMessage()`, giving a false sense of security.
**Suggested fix:**
```java
} catch (SdkException e) {
String errorMsg = e.getMessage();
boolean tokenLeaked = errorMsg != null && rawToken != null
&& errorMsg.contains(rawToken);
if (tokenLeaked) {
errorMsg = errorMsg.replace(rawToken, "[REDACTED]");
}
// SECURITY: If the token appeared in the STS error message, do not pass
// the original exception as-is -- its getMessage() still contains the token.
// Wrap with a sanitized message to prevent leakage via getCause().
Throwable cause = tokenLeaked
? new SdkException.Builder().message(errorMsg).cause(e.getCause()).build()
: e;
throw new CredentialResolutionException(
"Failed to assume role '" + cfg.roleArn
+ "' via AssumeRoleWithWebIdentity: " + errorMsg, cause);
}
```
And in the test, please add assertion on the cause:
```java
// In testTokenRedactedFromStsErrorMessage, add:
assertFalse(ex.getCause().getMessage().contains(TEST_TOKEN),
"Raw token must not leak via getCause().getMessage()");
```
##########
connector/credential-aws/src/main/java/org/apache/spark/security/aws/AwsStsCredentialProvider.java:
##########
@@ -29,18 +41,209 @@
/**
* :: DeveloperApi ::
- * AWS STS credential provider that exchanges an OIDC identity token for temporary
- * AWS credentials via {@code AssumeRoleWithWebIdentity}.
+ * A {@link CredentialProvider} that exchanges an OIDC identity token for temporary
+ * AWS credentials via the STS {@code AssumeRoleWithWebIdentity} API.
+ * <p>
+ * The returned {@link ServiceCredential} contains S3A-compatible Hadoop configuration
+ * properties ({@code fs.s3a.access.key}, {@code fs.s3a.secret.key},
+ * {@code fs.s3a.session.token}) that can be propagated to executors for accessing
+ * S3-compatible storage.
+ * <p>
+ * <b>Configuration keys</b> (passed via {@code spark.security.oidc.*}):
+ * <ul>
+ * <li>{@code spark.security.oidc.aws.roleArn} (required) — the ARN of the IAM role
+ * to assume</li>
+ * <li>{@code spark.security.oidc.aws.sessionName} (optional) — the role session name;
+ * defaults to a value derived from the user's principal or "spark-oidc"</li>
+ * <li>{@code spark.security.oidc.aws.durationSeconds} (optional) — credential duration
+ * in seconds (900–43200); if unset, STS uses the role's default maximum</li>
+ * <li>{@code spark.security.oidc.aws.region} (optional) — the AWS region for the STS
+ * endpoint; defaults to us-east-1 when only {@code stsEndpoint} is set</li>
+ * <li>{@code spark.security.oidc.aws.stsEndpoint} (optional) — a custom STS endpoint URL
+ * for non-AWS environments (MinIO, Ceph, LocalStack, etc.)</li>
+ * </ul>
+ * <p>
+ * <b>Security note:</b> The OIDC raw token is never included in log messages or
+ * exception messages. It is passed directly to the STS API call and discarded.
*
* @since 4.3.0
*/
@DeveloperApi
public class AwsStsCredentialProvider implements CredentialProvider {
+ // Configuration key constants
+ static final String CONF_ROLE_ARN = "spark.security.oidc.aws.roleArn";
+ static final String CONF_SESSION_NAME = "spark.security.oidc.aws.sessionName";
+ static final String CONF_DURATION_SECONDS = "spark.security.oidc.aws.durationSeconds";
+ static final String CONF_REGION = "spark.security.oidc.aws.region";
+ static final String CONF_STS_ENDPOINT = "spark.security.oidc.aws.stsEndpoint";
+
+ /** Minimum duration allowed by STS AssumeRoleWithWebIdentity (15 minutes). */
+ static final int MIN_DURATION_SECONDS = 900;
+ /** Maximum duration allowed by STS AssumeRoleWithWebIdentity (12 hours). */
+ static final int MAX_DURATION_SECONDS = 43200;
+
+ private static final String DEFAULT_REGION = "us-east-1";
+ private static final String DEFAULT_SESSION_NAME = "spark-oidc";
+
+ /**
+ * Immutable configuration holder that is safely published via the volatile
+ * {@link #config} field. All fields are set once during construction and are
+ * final, ensuring correct visibility across threads after init() completes.
+ */
+ static final class ResolvedConfig {
+ final String roleArn;
+ final String roleSessionName;
+ final Integer durationSeconds;
+ final Region resolvedRegion;
+ final URI endpointOverride;
+ final StsClient stsClient;
+
+ ResolvedConfig(String roleArn, String roleSessionName, Integer durationSeconds,
+ Region resolvedRegion, URI endpointOverride, StsClient stsClient) {
+ this.roleArn = roleArn;
+ this.roleSessionName = roleSessionName;
+ this.durationSeconds = durationSeconds;
+ this.resolvedRegion = resolvedRegion;
+ this.endpointOverride = endpointOverride;
+ this.stsClient = stsClient;
+ }
+ }
+
+ /** Safely published via volatile write in init(); read in resolve()/suggestedTtl(). */
+ private volatile ResolvedConfig config;
+
+ /**
+ * Default no-arg constructor used by {@link java.util.ServiceLoader}.
+ */
+ public AwsStsCredentialProvider() {
+ // ServiceLoader requires a public no-arg constructor
+ }
+
+ /**
+ * Package-private constructor for testing with an injected STS client.
+ * <p>
+ * This constructor is visible for testing only; production code must use the
+ * no-arg constructor followed by {@link #init(Map)}.
+ *
+ * @param stsClient the STS client to use (must not be null)
+ * @param roleArn the IAM role ARN (must not be null or blank)
+ * @param roleSessionName the session name (may be null for default)
+ * @param durationSeconds the credential duration in seconds (may be null)
+ */
+ @VisibleForTesting
+ AwsStsCredentialProvider(StsClient stsClient, String roleArn, String roleSessionName,
+ Integer durationSeconds) {
+ this.config = new ResolvedConfig(roleArn, roleSessionName, durationSeconds,
+ null, null, stsClient);
+ }
+
@Override
public void init(Map<String, String> conf) {
- throw new UnsupportedOperationException(
- "AwsStsCredentialProvider is a stub. Full implementation in SPARK-57898.");
+ if (this.config != null) {
+ throw new IllegalStateException("AwsStsCredentialProvider is already initialized");
+ }
+
+ String roleArn = conf.get(CONF_ROLE_ARN);
+ if (roleArn == null || roleArn.isBlank()) {
+ throw new IllegalArgumentException(
+ "Configuration key '" + CONF_ROLE_ARN + "' is required but was not set. "
+ + "Specify the ARN of the IAM role to assume via AssumeRoleWithWebIdentity.");
+ }
+
+ String roleSessionName = conf.get(CONF_SESSION_NAME);
+ String regionStr = conf.get(CONF_REGION);
+ String stsEndpoint = conf.get(CONF_STS_ENDPOINT);
+
+ Integer durationSeconds = null;
+ String durationStr = conf.get(CONF_DURATION_SECONDS);
+ if (durationStr != null && !durationStr.isBlank()) {
+ try {
+ durationSeconds = Integer.parseInt(durationStr.trim());
+ } catch (NumberFormatException e) {
+ throw new IllegalArgumentException(
+ "Configuration key '" + CONF_DURATION_SECONDS
+ + "' must be a valid integer, got: " + durationStr, e);
+ }
+ if (durationSeconds < MIN_DURATION_SECONDS || durationSeconds > MAX_DURATION_SECONDS) {
+ throw new IllegalArgumentException(
+ "Configuration key '" + CONF_DURATION_SECONDS + "' must be between "
+ + MIN_DURATION_SECONDS + " and " + MAX_DURATION_SECONDS
+ + " seconds, got: " + durationSeconds);
+ }
+ }
+
+ // Resolve the region and endpoint before building the client
+ Region resolvedRegion = resolveRegion(regionStr, stsEndpoint);
+ URI endpointOverride = resolveEndpoint(stsEndpoint);
+
+ StsClient stsClient = buildStsClient(resolvedRegion, endpointOverride);
+
+ // Single volatile write publishes all configuration atomically
+ this.config = new ResolvedConfig(roleArn, roleSessionName, durationSeconds,
+ resolvedRegion, endpointOverride, stsClient);
+ }
+
+ @Override
+ public void close() {
+ ResolvedConfig cfg = this.config;
+ if (cfg != null && cfg.stsClient != null) {
+ cfg.stsClient.close();
+ }
+ }
+
+ /**
+ * Resolves the AWS region based on explicit configuration and endpoint presence.
+ * When a custom endpoint is provided without an explicit region, defaults to us-east-1.
+ */
+ private static Region resolveRegion(String regionStr, String stsEndpoint) {
+ if (regionStr != null && !regionStr.isBlank()) {
+ return Region.of(regionStr);
+ } else if (stsEndpoint != null && !stsEndpoint.isBlank()) {
+ // When a custom endpoint is set but no explicit region, use a default region.
+ // The region is required by the SDK but not meaningful for non-AWS endpoints.
+ return Region.of(DEFAULT_REGION);
+ }
+ return null;
+ }
+
+ /**
+ * Resolves the endpoint override URI from configuration.
+ */
+ private static URI resolveEndpoint(String stsEndpoint) {
+ if (stsEndpoint != null && !stsEndpoint.isBlank()) {
+ return URI.create(stsEndpoint);
Review Comment:
If the user provides a malformed endpoint (e.g., `"not a valid uri"`), `URI.create()` throws a bare `IllegalArgumentException` with a message like `"Illegal character in scheme name"`. No mention of which config key caused the issue. All other config validation errors in `init()` include the key name for debuggability.
Suggested:
```java
private static URI resolveEndpoint(String stsEndpoint) {
if (stsEndpoint != null && !stsEndpoint.isBlank()) {
try {
return URI.create(stsEndpoint.trim());
} catch (IllegalArgumentException e) {
throw new IllegalArgumentException(
"Configuration key '" + CONF_STS_ENDPOINT
+ "' must be a valid URI, got: " + stsEndpoint, e);
}
}
return null;
}
```
##########
connector/credential-aws/src/main/java/org/apache/spark/security/aws/AwsStsCredentialProvider.java:
##########
@@ -29,18 +41,209 @@
/**
* :: DeveloperApi ::
- * AWS STS credential provider that exchanges an OIDC identity token for temporary
- * AWS credentials via {@code AssumeRoleWithWebIdentity}.
+ * A {@link CredentialProvider} that exchanges an OIDC identity token for temporary
+ * AWS credentials via the STS {@code AssumeRoleWithWebIdentity} API.
+ * <p>
+ * The returned {@link ServiceCredential} contains S3A-compatible Hadoop configuration
+ * properties ({@code fs.s3a.access.key}, {@code fs.s3a.secret.key},
+ * {@code fs.s3a.session.token}) that can be propagated to executors for accessing
+ * S3-compatible storage.
+ * <p>
+ * <b>Configuration keys</b> (passed via {@code spark.security.oidc.*}):
+ * <ul>
+ * <li>{@code spark.security.oidc.aws.roleArn} (required) — the ARN of the IAM role
+ * to assume</li>
+ * <li>{@code spark.security.oidc.aws.sessionName} (optional) — the role session name;
+ * defaults to a value derived from the user's principal or "spark-oidc"</li>
+ * <li>{@code spark.security.oidc.aws.durationSeconds} (optional) — credential duration
+ * in seconds (900–43200); if unset, STS uses the role's default maximum</li>
+ * <li>{@code spark.security.oidc.aws.region} (optional) — the AWS region for the STS
+ * endpoint; defaults to us-east-1 when only {@code stsEndpoint} is set</li>
+ * <li>{@code spark.security.oidc.aws.stsEndpoint} (optional) — a custom STS endpoint URL
+ * for non-AWS environments (MinIO, Ceph, LocalStack, etc.)</li>
+ * </ul>
+ * <p>
+ * <b>Security note:</b> The OIDC raw token is never included in log messages or
+ * exception messages. It is passed directly to the STS API call and discarded.
*
* @since 4.3.0
*/
@DeveloperApi
public class AwsStsCredentialProvider implements CredentialProvider {
+ // Configuration key constants
+ static final String CONF_ROLE_ARN = "spark.security.oidc.aws.roleArn";
+ static final String CONF_SESSION_NAME = "spark.security.oidc.aws.sessionName";
+ static final String CONF_DURATION_SECONDS = "spark.security.oidc.aws.durationSeconds";
+ static final String CONF_REGION = "spark.security.oidc.aws.region";
+ static final String CONF_STS_ENDPOINT = "spark.security.oidc.aws.stsEndpoint";
+
+ /** Minimum duration allowed by STS AssumeRoleWithWebIdentity (15 minutes). */
+ static final int MIN_DURATION_SECONDS = 900;
+ /** Maximum duration allowed by STS AssumeRoleWithWebIdentity (12 hours). */
+ static final int MAX_DURATION_SECONDS = 43200;
+
+ private static final String DEFAULT_REGION = "us-east-1";
+ private static final String DEFAULT_SESSION_NAME = "spark-oidc";
+
+ /**
+ * Immutable configuration holder that is safely published via the volatile
+ * {@link #config} field. All fields are set once during construction and are
+ * final, ensuring correct visibility across threads after init() completes.
+ */
+ static final class ResolvedConfig {
+ final String roleArn;
+ final String roleSessionName;
+ final Integer durationSeconds;
+ final Region resolvedRegion;
+ final URI endpointOverride;
+ final StsClient stsClient;
+
+ ResolvedConfig(String roleArn, String roleSessionName, Integer durationSeconds,
+ Region resolvedRegion, URI endpointOverride, StsClient stsClient) {
+ this.roleArn = roleArn;
+ this.roleSessionName = roleSessionName;
+ this.durationSeconds = durationSeconds;
+ this.resolvedRegion = resolvedRegion;
+ this.endpointOverride = endpointOverride;
+ this.stsClient = stsClient;
+ }
+ }
+
+ /** Safely published via volatile write in init(); read in resolve()/suggestedTtl(). */
+ private volatile ResolvedConfig config;
+
+ /**
+ * Default no-arg constructor used by {@link java.util.ServiceLoader}.
+ */
+ public AwsStsCredentialProvider() {
+ // ServiceLoader requires a public no-arg constructor
+ }
+
+ /**
+ * Package-private constructor for testing with an injected STS client.
+ * <p>
+ * This constructor is visible for testing only; production code must use the
+ * no-arg constructor followed by {@link #init(Map)}.
+ *
+ * @param stsClient the STS client to use (must not be null)
+ * @param roleArn the IAM role ARN (must not be null or blank)
+ * @param roleSessionName the session name (may be null for default)
+ * @param durationSeconds the credential duration in seconds (may be null)
+ */
+ @VisibleForTesting
+ AwsStsCredentialProvider(StsClient stsClient, String roleArn, String roleSessionName,
+ Integer durationSeconds) {
+ this.config = new ResolvedConfig(roleArn, roleSessionName, durationSeconds,
+ null, null, stsClient);
+ }
+
@Override
public void init(Map<String, String> conf) {
- throw new UnsupportedOperationException(
- "AwsStsCredentialProvider is a stub. Full implementation in SPARK-57898.");
+ if (this.config != null) {
+ throw new IllegalStateException("AwsStsCredentialProvider is already initialized");
+ }
+
+ String roleArn = conf.get(CONF_ROLE_ARN);
+ if (roleArn == null || roleArn.isBlank()) {
+ throw new IllegalArgumentException(
+ "Configuration key '" + CONF_ROLE_ARN + "' is required but was not set. "
+ + "Specify the ARN of the IAM role to assume via AssumeRoleWithWebIdentity.");
+ }
+
+ String roleSessionName = conf.get(CONF_SESSION_NAME);
+ String regionStr = conf.get(CONF_REGION);
+ String stsEndpoint = conf.get(CONF_STS_ENDPOINT);
+
+ Integer durationSeconds = null;
+ String durationStr = conf.get(CONF_DURATION_SECONDS);
+ if (durationStr != null && !durationStr.isBlank()) {
+ try {
+ durationSeconds = Integer.parseInt(durationStr.trim());
Review Comment:
`durationStr` is explicitly trimmed before parsing (`durationStr.trim()`), but `roleArn`, `regionStr`, and `stsEndpoint` are not trimmed. Configuration values commonly pick up stray whitespace from YAML or properties files, and a trailing space in a role ARN (e.g., `"arn:aws:iam::123:role/test "`) would pass the blank check but cause a cryptic STS `MalformedPolicyDocumentException`.
Should we apply consistent trimming to all config values? e.g.:
```java
String roleArn = conf.get(CONF_ROLE_ARN);
if (roleArn != null) roleArn = roleArn.trim();
if (roleArn == null || roleArn.isEmpty()) {
throw new IllegalArgumentException(...);
}
// Similarly for regionStr and stsEndpoint
```
##########
connector/credential-aws/src/main/java/org/apache/spark/security/aws/AwsStsCredentialProvider.java:
##########
@@ -29,18 +41,209 @@
/**
* :: DeveloperApi ::
- * AWS STS credential provider that exchanges an OIDC identity token for temporary
- * AWS credentials via {@code AssumeRoleWithWebIdentity}.
+ * A {@link CredentialProvider} that exchanges an OIDC identity token for temporary
+ * AWS credentials via the STS {@code AssumeRoleWithWebIdentity} API.
+ * <p>
+ * The returned {@link ServiceCredential} contains S3A-compatible Hadoop configuration
+ * properties ({@code fs.s3a.access.key}, {@code fs.s3a.secret.key},
+ * {@code fs.s3a.session.token}) that can be propagated to executors for accessing
+ * S3-compatible storage.
+ * <p>
+ * <b>Configuration keys</b> (passed via {@code spark.security.oidc.*}):
+ * <ul>
+ * <li>{@code spark.security.oidc.aws.roleArn} (required) — the ARN of the IAM role
Review Comment:
A few documentation nits:
1. The config key descriptions contain non-ASCII characters:
- Em dash (U+2014) in 5 places (e.g., `(required) — the ARN`)
- En dash (U+2013) in `900–43200`
Could you replace them all with `--`?
2. Consider adding a note about the behavior when neither `region` nor `stsEndpoint` is configured -- the SDK's default region resolution (AWS_REGION env var, ~/.aws/config, etc.) is used.
##########
connector/credential-aws/src/main/java/org/apache/spark/security/aws/AwsStsCredentialProvider.java:
##########
@@ -29,18 +41,209 @@
/**
* :: DeveloperApi ::
- * AWS STS credential provider that exchanges an OIDC identity token for temporary
- * AWS credentials via {@code AssumeRoleWithWebIdentity}.
+ * A {@link CredentialProvider} that exchanges an OIDC identity token for temporary
+ * AWS credentials via the STS {@code AssumeRoleWithWebIdentity} API.
+ * <p>
+ * The returned {@link ServiceCredential} contains S3A-compatible Hadoop configuration
+ * properties ({@code fs.s3a.access.key}, {@code fs.s3a.secret.key},
+ * {@code fs.s3a.session.token}) that can be propagated to executors for accessing
+ * S3-compatible storage.
+ * <p>
+ * <b>Configuration keys</b> (passed via {@code spark.security.oidc.*}):
+ * <ul>
+ * <li>{@code spark.security.oidc.aws.roleArn} (required) — the ARN of the IAM role
+ * to assume</li>
+ * <li>{@code spark.security.oidc.aws.sessionName} (optional) — the role session name;
+ * defaults to a value derived from the user's principal or "spark-oidc"</li>
+ * <li>{@code spark.security.oidc.aws.durationSeconds} (optional) — credential duration
+ * in seconds (900–43200); if unset, STS uses the role's default maximum</li>
+ * <li>{@code spark.security.oidc.aws.region} (optional) — the AWS region for the STS
+ * endpoint; defaults to us-east-1 when only {@code stsEndpoint} is set</li>
+ * <li>{@code spark.security.oidc.aws.stsEndpoint} (optional) — a custom STS endpoint URL
+ * for non-AWS environments (MinIO, Ceph, LocalStack, etc.)</li>
+ * </ul>
+ * <p>
+ * <b>Security note:</b> The OIDC raw token is never included in log messages or
+ * exception messages. It is passed directly to the STS API call and discarded.
*
* @since 4.3.0
*/
@DeveloperApi
Review Comment:
I know this was present from the scaffolding PR, but since you're touching this file, could you remove `@DeveloperApi` (and the `:: DeveloperApi ::` in the Javadoc) because `AwsStsCredentialProvider` is an implementation of `CredentialProvider` SPI ?
##########
connector/credential-aws/src/main/java/org/apache/spark/security/aws/AwsStsCredentialProvider.java:
##########
@@ -51,7 +254,102 @@ public Set<String> supportedSchemes() {
@Override
public ServiceCredential resolve(UserContext user, URI target)
throws CredentialResolutionException {
- throw new UnsupportedOperationException(
- "AwsStsCredentialProvider is a stub. Full implementation in SPARK-57898.");
+ if (user == null) {
+ throw new CredentialResolutionException(
+ "UserContext must not be null when resolving AWS credentials");
+ }
+ if (target == null) {
+ throw new CredentialResolutionException(
+ "Target URI must not be null when resolving AWS credentials");
+ }
+ if (user.getRawToken() == null || user.getRawToken().isBlank()) {
Review Comment:
nit: `user.getRawToken()` is called twice in the null/blank check before it is assigned to a local variable. While `UserContext` is immutable, consolidating to a single call at the top is more defensive and consistent:
```java
String rawToken = user.getRawToken();
if (rawToken == null || rawToken.isBlank()) {
throw new CredentialResolutionException(
"UserContext raw token must not be null or blank; cannot perform "
+ "AssumeRoleWithWebIdentity without an identity token");
}
```
##########
connector/credential-aws/src/main/java/org/apache/spark/security/aws/AwsStsCredentialProvider.java:
##########
@@ -51,7 +254,102 @@ public Set<String> supportedSchemes() {
@Override
public ServiceCredential resolve(UserContext user, URI target)
throws CredentialResolutionException {
- throw new UnsupportedOperationException(
- "AwsStsCredentialProvider is a stub. Full implementation in SPARK-57898.");
+ if (user == null) {
+ throw new CredentialResolutionException(
+ "UserContext must not be null when resolving AWS credentials");
+ }
+ if (target == null) {
+ throw new CredentialResolutionException(
+ "Target URI must not be null when resolving AWS credentials");
+ }
+ if (user.getRawToken() == null || user.getRawToken().isBlank()) {
+ throw new CredentialResolutionException(
+ "UserContext raw token must not be null or blank; cannot perform "
+ + "AssumeRoleWithWebIdentity without an identity token");
+ }
+
+ ResolvedConfig cfg = this.config;
+ if (cfg == null) {
+ throw new CredentialResolutionException("resolve() called before init()");
+ }
+ String sessionName = cfg.roleSessionName;
+ if (sessionName == null || sessionName.isBlank()) {
+ // Derive from principal, sanitizing for STS session name constraints
+ // (alphanumeric, =,.@- only, max 64 chars)
+ String principal = user.getPrincipal();
+ if (principal != null && !principal.isBlank()) {
+ sessionName = sanitizeSessionName(principal);
+ } else {
+ sessionName = DEFAULT_SESSION_NAME;
+ }
+ }
+
+ String rawToken = user.getRawToken();
+ try {
+ AssumeRoleWithWebIdentityRequest.Builder reqBuilder =
+ AssumeRoleWithWebIdentityRequest.builder()
+ .roleArn(cfg.roleArn)
+ .roleSessionName(sessionName)
+ .webIdentityToken(rawToken);
+
+ if (cfg.durationSeconds != null) {
+ reqBuilder.durationSeconds(cfg.durationSeconds);
+ }
+
+ AssumeRoleWithWebIdentityResponse response =
+ cfg.stsClient.assumeRoleWithWebIdentity(reqBuilder.build());
+
+ Credentials creds = response.credentials();
+ if (creds == null || creds.accessKeyId() == null
+ || creds.secretAccessKey() == null || creds.sessionToken() == null) {
+ throw new CredentialResolutionException(
+ "STS returned incomplete credentials for role '" + cfg.roleArn + "'");
+ }
+
+ Map<String, String> properties = Map.of(
+ "fs.s3a.access.key", creds.accessKeyId(),
+ "fs.s3a.secret.key", creds.secretAccessKey(),
+ "fs.s3a.session.token", creds.sessionToken()
+ );
+
+ Instant expiration = creds.expiration();
+ return new ServiceCredential(properties, expiration);
+ } catch (SdkException e) {
+ // SECURITY: Never include the token in exception messages.
+ // Defensively strip any occurrence of the raw token from the STS error message
+ // in case the service accidentally echoed it.
+ String errorMsg = e.getMessage();
+ if (errorMsg != null && rawToken != null && errorMsg.contains(rawToken)) {
+ errorMsg = errorMsg.replace(rawToken, "[REDACTED]");
+ }
+ throw new CredentialResolutionException(
+ "Failed to assume role '" + cfg.roleArn + "' via AssumeRoleWithWebIdentity: "
+ + errorMsg, e);
+ }
+ }
+
+ @Override
+ public Duration suggestedTtl() {
+ ResolvedConfig cfg = this.config;
+ if (cfg != null && cfg.durationSeconds != null) {
+ return Duration.ofSeconds(cfg.durationSeconds);
+ }
+ return Duration.ofMinutes(15);
+ }
+
+ /**
+ * Sanitizes a principal string to be valid as an STS role session name.
+ * STS session names must match [\w+=,.@-]{2,64}. We use an explicit character
+ * class rather than {@code \w} to avoid locale-dependent behavior.
+ */
+ static String sanitizeSessionName(String principal) {
+ String sanitized = principal.replaceAll("[^a-zA-Z0-9_+=,.@\\-]", "-");
Review Comment:
`String.replaceAll()` compiles the regex on every call. Since `sanitizeSessionName` is called from `resolve()` (which may be invoked frequently), caching the compiled `Pattern` as a static field avoids repeated compilation:
```java
private static final Pattern INVALID_SESSION_CHARS =
Pattern.compile("[^a-zA-Z0-9_+=,.@\\-]");
static String sanitizeSessionName(String principal) {
String sanitized = INVALID_SESSION_CHARS.matcher(principal).replaceAll("-");
...
}
```
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57898][CORE] Implement AwsStsCredentialProvider (OIDC to STS credential exchange) [spark]
Posted by "sarutak (via GitHub)" <gi...@apache.org>.
sarutak commented on code in PR #57655:
URL: https://github.com/apache/spark/pull/57655#discussion_r3688473588
##########
pom.xml:
##########
@@ -3519,6 +3519,13 @@
</modules>
</profile>
+ <profile>
+ <id>credential-aws</id>
+ <modules>
+ <module>connector/credential-aws</module>
Review Comment:
Since introducing a new Maven profile requires updates to many CI/release/lint scripts, I'd like to split the module scaffolding (SPARK-57897) into a separate PR as originally planed. I'll handle it. It will include:
- Module definition (`connector/credential-aws/pom.xml` with a stub provider)
- `pom.xml` profile, `SparkBuild.scala`, `modules.py`, `utils.py`
- `META-INF/services` registration
- CI/release/lint script updates:
- `.github/workflows/build_and_test.yml`
- `.github/workflows/maven_test.yml`
- `.github/workflows/python_hosted_runner_test.yml`
- `.github/workflows/benchmark.yml`
- `dev/create-release/release-build.sh`
- `dev/test-dependencies.sh`
- `dev/mima`
- `dev/lint-java`
- `dev/sbt-checkstyle`
- `dev/scalastyle`
- `docs/_plugins/build_api_docs.rb`
- `dev/spark-test-image-util/docs/build-docs`
Once that's merged, this PR can focus purely on the `AwsStsCredentialProvider` implementation and tests, removing the build wiring changes from its diff.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57898][CORE] Implement AwsStsCredentialProvider (OIDC to STS credential exchange) [spark]
Posted by "yadavay-amzn (via GitHub)" <gi...@apache.org>.
yadavay-amzn commented on PR #57655:
URL: https://github.com/apache/spark/pull/57655#issuecomment-5207633341
@sarutak @dongjoon-hyun @HyukjinKwon Thanks for reviewing! Addressed comments in the latest commit, please take another look when you get a chance. Thanks again!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org