This is an automated email from the ASF dual-hosted git repository.
sarutak pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 3f598f070f91 [SPARK-57898][CORE] Implement AwsStsCredentialProvider
(OIDC to STS credential exchange)
3f598f070f91 is described below
commit 3f598f070f91b1175f405f07f5aec427a8221bc5
Author: Anupam Yadav <[email protected]>
AuthorDate: Wed Aug 12 20:31:37 2026 +0900
[SPARK-57898][CORE] Implement AwsStsCredentialProvider (OIDC to STS
credential exchange)
### 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 as a
`ServiceCredential`.
- Configuration under `spark.security.oidc.aws.*`: `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()`.
- Implements `close()` (SPI extends `AutoCloseable`) to release the STS
client's HTTP resources; tests exercise resolve-after-close.
- Never exposes the OIDC raw token: the top-level exception message and the
entire `SdkException` cause chain are scanned and redacted, with the original
cause dropped if the token appeared anywhere in the chain.
Notes for reviewers:
- The class does NOT carry `DeveloperApi`. This is intentional:
`AwsStsCredentialProvider` is a ServiceLoader-discovered implementation of the
SPI, not itself a public developer API. The SPI (`CredentialProvider`) carries
`DeveloperApi`; end users do not program against this class directly. `since
4.4.0` is kept in the Javadoc.
- User-facing documentation for the new `spark.security.oidc.aws.*`
configuration keys is not part of this PR. It will land under
[SPARK-57901](https://issues.apache.org/jira/browse/SPARK-57901) (the OIDC
SPIP's documentation subtask) alongside the other `spark.security.oidc.*` keys
introduced by tasks 1-4.
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.*` AWS
configuration keys.
### How was this patch tested?
New `AwsStsCredentialProviderSuite` (JUnit 5 + Mockito) covers:
- Successful resolve with `ArgumentCaptor` assertions on the STS request
(role ARN, web identity token, session name, duration) and on the mapped
`fs.s3a.*` keys plus expiration.
- Token safety: STS failure wrapped in `CredentialResolutionException`;
asserts the raw token never appears in the top-level message; adversarial test
with the token planted inside a nested cause verifies the full cause chain is
scrubbed.
- Incomplete STS credentials (null credentials or missing session token)
throw `CredentialResolutionException` rather than NPE.
- Required-config / precondition checks: missing/blank `roleArn`,
null/blank token, null `target`, `resolve()` before `init()`, re-initialization
guard.
- `durationSeconds` range validation (900-43200 inclusive, plus zero,
negative, and non-numeric).
- `sessionName` validation against the STS `[a-zA-Z0-9_+=,.-]{2,64}`
character/length rules; trimming for `roleArn`, `region`, `stsEndpoint`, and
`sessionName`.
- Malformed `stsEndpoint` throws `IllegalArgumentException` naming the
config key.
- Resolved region/endpoint for endpoint+region, endpoint-only (default
region), and neither.
- Session-name derivation from principal (including sanitization edge
cases: truncation over 64 chars, invalid-character replacement, backslash
handling).
- `suggestedTtl`, `supportedSchemes`, `ServiceLoader` discovery.
- `close()` releases the STS client; `resolve()` after `close()` is wrapped
in `CredentialResolutionException` rather than surfacing a raw
`IllegalStateException`.
- Test suite is region-environment-independent (sets and restores
`aws.region` at suite scope) so it passes on GHA runners without any AWS
environment configured.
### Was this patch authored or co-authored using generative AI tooling?
Yes, authored with assistance from Claude Opus 4.8
Closes #57655 from yadavay-amzn/SPARK-57898.
Authored-by: Anupam Yadav <[email protected]>
Signed-off-by: Kousuke Saruta <[email protected]>
---
connector/credential-aws/pom.xml | 10 +
.../security/aws/AwsStsCredentialProvider.java | 425 ++++++++-
.../aws/AwsStsCredentialProviderSuite.java | 955 ++++++++++++++++++++-
3 files changed, 1368 insertions(+), 22 deletions(-)
diff --git a/connector/credential-aws/pom.xml b/connector/credential-aws/pom.xml
index b1cdf5c58f37..e32666347659 100644
--- a/connector/credential-aws/pom.xml
+++ b/connector/credential-aws/pom.xml
@@ -48,6 +48,11 @@
<groupId>org.apache.spark</groupId>
<artifactId>spark-tags_${scala.binary.version}</artifactId>
</dependency>
+ <dependency>
+ <groupId>software.amazon.awssdk</groupId>
+ <artifactId>sts</artifactId>
+ <version>${aws.java.sdk.v2.version}</version>
+ </dependency>
<!-- Test dependencies -->
<!--
@@ -65,6 +70,11 @@
<artifactId>junit-jupiter</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-core</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
diff --git
a/connector/credential-aws/src/main/java/org/apache/spark/security/aws/AwsStsCredentialProvider.java
b/connector/credential-aws/src/main/java/org/apache/spark/security/aws/AwsStsCredentialProvider.java
index 07d8821b6569..f8b967d1fd83 100644
---
a/connector/credential-aws/src/main/java/org/apache/spark/security/aws/AwsStsCredentialProvider.java
+++
b/connector/credential-aws/src/main/java/org/apache/spark/security/aws/AwsStsCredentialProvider.java
@@ -18,29 +18,289 @@
package org.apache.spark.security.aws;
import java.net.URI;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Collections;
+import java.util.IdentityHashMap;
import java.util.Map;
import java.util.Set;
+import java.util.regex.Pattern;
+import java.util.stream.Stream;
+
+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
+ * @since 4.4.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 [a-zA-Z0-9_+=,.@-]{2,64}} per AWS STS documentation.
+ * Uses an explicit ASCII character class rather than {@code \w} to avoid
+ * accepting non-ASCII characters (e.g. accented letters, CJK) that AWS STS
+ * would reject.
+ */
+ private static final Pattern SESSION_NAME_VALID_PATTERN =
+ Pattern.compile("[a-zA-Z0-9_+=,.@\\-]{2,64}");
+
+ /**
+ * 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;
+
+ /** Guards against double-close and allows resolve() to fail fast after
close(). */
+ private volatile boolean closed = false;
+
+ /**
+ * 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)
+ */
+ // Visible for testing only.
+ 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);
+ if (roleSessionName != null) {
+ roleSessionName = roleSessionName.trim();
+ }
+ if (roleSessionName != null && roleSessionName.isBlank()) {
+ roleSessionName = null;
+ }
+ if (roleSessionName != null &&
!SESSION_NAME_VALID_PATTERN.matcher(roleSessionName).matches()) {
+ throw new IllegalArgumentException(
+ "Configuration key '" + CONF_SESSION_NAME
+ + "' must match [a-zA-Z0-9_+=,.@-]{2,64}, got: "
+ + roleSessionName);
+ }
+ String regionStr = conf.get(CONF_REGION);
+ if (regionStr != null) {
+ regionStr = regionStr.trim();
+ }
+ String stsEndpoint = conf.get(CONF_STS_ENDPOINT);
+ if (stsEndpoint != null) {
+ stsEndpoint = stsEndpoint.trim();
+ }
+
+ 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() {
+ if (closed) {
+ return;
+ }
+ closed = true;
+ 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.
+ * When neither region nor stsEndpoint is configured, returns null so the
STS client
+ * falls back to the AWS SDK default region resolution (AWS_REGION /
AWS_DEFAULT_REGION
+ * environment variables, then the ~/.aws/config profile).
+ */
+ 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()) {
+ try {
+ return URI.create(stsEndpoint);
+ } catch (IllegalArgumentException e) {
+ throw new IllegalArgumentException(
+ "Configuration key '" + CONF_STS_ENDPOINT + "' contains a
malformed URI: "
+ + stsEndpoint, e);
+ }
+ }
+ return null;
+ }
+
+ /**
+ * Builds the STS client with the resolved region and endpoint.
+ */
+ private static StsClient buildStsClient(Region resolvedRegion, URI
endpointOverride) {
+ StsClientBuilder builder = StsClient.builder()
+ // AssumeRoleWithWebIdentity does not require AWS credentials;
+ // the OIDC token itself serves as the authentication mechanism.
+ .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.
+ */
+ // Visible for testing only.
+ ResolvedConfig resolvedConfig() {
+ return config;
}
@Override
@@ -51,7 +311,152 @@ public class AwsStsCredentialProvider implements
CredentialProvider {
@Override
public ServiceCredential resolve(UserContext user, URI target)
throws CredentialResolutionException {
- throw new UnsupportedOperationException(
- "AwsStsCredentialProvider is a stub. Full implementation in
SPARK-57898.");
+ if (closed) {
+ throw new CredentialResolutionException("provider is closed");
+ }
+ 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");
+ }
+ 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");
+ }
+
+ 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(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();
+ String redactedMsg = errorMsg;
+ if (redactedMsg != null && rawToken != null &&
redactedMsg.contains(rawToken)) {
+ redactedMsg = redactedMsg.replace(rawToken, "[REDACTED]");
+ }
+ // Walk the entire cause chain to check if the token leaked into any
layer.
+ boolean tokenInAnyMessage = causeChainContainsToken(e, rawToken);
+ Throwable cause;
+ if (tokenInAnyMessage || (errorMsg != null &&
errorMsg.contains(rawToken))) {
+ // Drop the original cause chain entirely; replace with a sanitized
wrapper.
+ cause = SdkException.builder()
+ .message(redactedMsg)
+ .build();
+ } else {
+ cause = e;
+ }
+ throw new CredentialResolutionException(
+ "Failed to assume role '" + cfg.roleArn + "' via
AssumeRoleWithWebIdentity: "
+ + redactedMsg, cause);
+ } catch (IllegalStateException e) {
+ // The SDK throws IllegalStateException when the client has been closed.
+ throw new CredentialResolutionException(
+ "Failed to assume role '" + cfg.roleArn + "' via
AssumeRoleWithWebIdentity: "
+ + "the STS client has been closed", e);
+ }
+ }
+
+ @Override
+ public Duration suggestedTtl() {
+ ResolvedConfig cfg = this.config;
+ if (cfg != null && cfg.durationSeconds != null) {
+ return Duration.ofSeconds(cfg.durationSeconds);
+ }
+ return Duration.ofMinutes(15);
+ }
+
+ /**
+ * Walks the cause chain of the given throwable and checks whether the raw
token
+ * appears in any layer's message. Used to decide whether the original
exception
+ * chain is safe to preserve as a cause.
+ */
+ private static boolean causeChainContainsToken(Throwable root, String
rawToken) {
+ if (rawToken == null) {
+ return false;
+ }
+ return causeChainStream(root)
+ .anyMatch(t -> t.getMessage() != null &&
t.getMessage().contains(rawToken));
+ }
+
+ /**
+ * Returns a sequential stream over the cause chain starting from {@code
root}.
+ * Uses identity-based cycle detection to guard against circular cause
chains.
+ */
+ private static Stream<Throwable> causeChainStream(Throwable root) {
+ Stream.Builder<Throwable> builder = Stream.builder();
+ Set<Throwable> visited = Collections.newSetFromMap(new
IdentityHashMap<>());
+ Throwable current = root;
+ while (current != null && visited.add(current)) {
+ builder.accept(current);
+ current = current.getCause();
+ }
+ return builder.build();
+ }
+
+ /**
+ * Sanitizes a principal string to be valid as an STS role session name.
+ * STS session names must match [a-zA-Z0-9_+=,.@-]{2,64}. Both the validation
+ * pattern and the sanitization replacement use an explicit ASCII character
class
+ * rather than {@code \w} to avoid locale-dependent behavior and to reject
+ * non-ASCII characters that AWS STS would not accept.
+ */
+ static String sanitizeSessionName(String principal) {
+ String sanitized =
SESSION_NAME_INVALID_CHARS.matcher(principal).replaceAll("-");
+ if (sanitized.length() > 64) {
+ sanitized = sanitized.substring(0, 64);
+ }
+ if (sanitized.length() < 2) {
+ return DEFAULT_SESSION_NAME;
+ }
+ return sanitized;
}
}
diff --git
a/connector/credential-aws/src/test/java/org/apache/spark/security/aws/AwsStsCredentialProviderSuite.java
b/connector/credential-aws/src/test/java/org/apache/spark/security/aws/AwsStsCredentialProviderSuite.java
index cab91fcae636..5f6daa102c8a 100644
---
a/connector/credential-aws/src/test/java/org/apache/spark/security/aws/AwsStsCredentialProviderSuite.java
+++
b/connector/credential-aws/src/test/java/org/apache/spark/security/aws/AwsStsCredentialProviderSuite.java
@@ -19,22 +19,89 @@ package org.apache.spark.security.aws;
import java.net.URI;
import java.time.Duration;
+import java.time.Instant;
+import java.util.HashMap;
import java.util.Map;
import java.util.ServiceLoader;
import java.util.Set;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
-import static org.junit.jupiter.api.Assertions.*;
+import org.mockito.ArgumentCaptor;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.sts.StsClient;
+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 software.amazon.awssdk.services.sts.model.StsException;
import org.apache.spark.security.CredentialProvider;
+import org.apache.spark.security.CredentialResolutionException;
+import org.apache.spark.security.ServiceCredential;
+import org.apache.spark.security.UserContext;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
/**
- * Tests that {@link AwsStsCredentialProvider} is discoverable via {@link
ServiceLoader}.
+ * Tests for {@link AwsStsCredentialProvider}.
*/
public class AwsStsCredentialProviderSuite {
+ private static String previousAwsRegion;
+
+ /**
+ * Set the aws.region system property so that the AWS SDK's
+ * DefaultAwsRegionProviderChain can resolve a region on any environment
+ * (including CI runners with no AWS configuration). This does NOT affect
+ * AwsStsCredentialProvider.resolveRegion() which reads only from the conf
Map.
+ */
+ @BeforeAll
+ static void setUpClass() {
+ previousAwsRegion = System.getProperty("aws.region");
+ System.setProperty("aws.region", "us-east-1");
+ }
+
+ @AfterAll
+ static void tearDownClass() {
+ if (previousAwsRegion == null) {
+ System.clearProperty("aws.region");
+ } else {
+ System.setProperty("aws.region", previousAwsRegion);
+ }
+ }
+
+ /** Suite-level field closed by tearDown to avoid resource leaks from real
StsClients. */
+ private AwsStsCredentialProvider provider;
+
+ @AfterEach
+ void tearDown() {
+ if (provider != null) {
+ provider.close();
+ provider = null;
+ }
+ }
+
+ private static final String TEST_ROLE_ARN =
"arn:aws:iam::123456789012:role/test-role";
+ private static final String TEST_TOKEN =
"eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.test-payload";
+ private static final String TEST_PRINCIPAL = "[email protected]";
+ private static final String TEST_ISSUER = "https://idp.example.com";;
+ private static final URI TEST_TARGET =
URI.create("s3a://my-bucket/data/file.parquet");
+
+ // ========== ServiceLoader Discovery ==========
+
@Test
- void testServiceLoaderDiscovery() {
+ public void testServiceLoaderDiscovery() {
ServiceLoader<CredentialProvider> loader =
ServiceLoader.load(CredentialProvider.class);
boolean found = false;
for (CredentialProvider provider : loader) {
@@ -46,29 +113,893 @@ public class AwsStsCredentialProviderSuite {
assertTrue(found, "AwsStsCredentialProvider should be discoverable via
ServiceLoader");
}
+ // ========== init() ==========
+
@Test
- void testSupportedSchemes() {
+ public void testMissingRoleArnThrowsIllegalArgumentException() {
+ Map<String, String> conf = new HashMap<>();
+
AwsStsCredentialProvider provider = new AwsStsCredentialProvider();
- Set<String> schemes = provider.supportedSchemes();
- assertEquals(Set.of("s3a"), schemes);
+ IllegalArgumentException ex = assertThrows(IllegalArgumentException.class,
+ () -> provider.init(conf));
+
assertTrue(ex.getMessage().contains(AwsStsCredentialProvider.CONF_ROLE_ARN));
+ }
+
+ @Test
+ public void testBlankRoleArnThrowsIllegalArgumentException() {
+ Map<String, String> conf = new HashMap<>();
+ conf.put(AwsStsCredentialProvider.CONF_ROLE_ARN, " ");
+
+ AwsStsCredentialProvider provider = new AwsStsCredentialProvider();
+ IllegalArgumentException ex = assertThrows(IllegalArgumentException.class,
+ () -> provider.init(conf));
+
assertTrue(ex.getMessage().contains(AwsStsCredentialProvider.CONF_ROLE_ARN));
+ }
+
+ @Test
+ public void testInitWithInvalidDurationSeconds() {
+ Map<String, String> conf = new HashMap<>();
+ conf.put(AwsStsCredentialProvider.CONF_ROLE_ARN, TEST_ROLE_ARN);
+ conf.put(AwsStsCredentialProvider.CONF_DURATION_SECONDS, "not-a-number");
+
+ AwsStsCredentialProvider provider = new AwsStsCredentialProvider();
+ assertThrows(IllegalArgumentException.class, () -> provider.init(conf));
+ }
+
+ @Test
+ public void testInitWithZeroDurationSecondsThrowsIllegalArgumentException() {
+ Map<String, String> conf = new HashMap<>();
+ conf.put(AwsStsCredentialProvider.CONF_ROLE_ARN, TEST_ROLE_ARN);
+ conf.put(AwsStsCredentialProvider.CONF_DURATION_SECONDS, "0");
+
+ AwsStsCredentialProvider provider = new AwsStsCredentialProvider();
+ IllegalArgumentException ex = assertThrows(IllegalArgumentException.class,
+ () -> provider.init(conf));
+
assertTrue(ex.getMessage().contains(AwsStsCredentialProvider.CONF_DURATION_SECONDS));
+ assertTrue(ex.getMessage().contains("900"));
+ assertTrue(ex.getMessage().contains("43200"));
+ }
+
+ @Test
+ public void
testInitWithNegativeDurationSecondsThrowsIllegalArgumentException() {
+ Map<String, String> conf = new HashMap<>();
+ conf.put(AwsStsCredentialProvider.CONF_ROLE_ARN, TEST_ROLE_ARN);
+ conf.put(AwsStsCredentialProvider.CONF_DURATION_SECONDS, "-100");
+
+ AwsStsCredentialProvider provider = new AwsStsCredentialProvider();
+ IllegalArgumentException ex = assertThrows(IllegalArgumentException.class,
+ () -> provider.init(conf));
+
assertTrue(ex.getMessage().contains(AwsStsCredentialProvider.CONF_DURATION_SECONDS));
+ assertTrue(ex.getMessage().contains("900"));
+ }
+
+ @Test
+ public void testInitWithDurationBelowMinimumThrowsIllegalArgumentException()
{
+ Map<String, String> conf = new HashMap<>();
+ conf.put(AwsStsCredentialProvider.CONF_ROLE_ARN, TEST_ROLE_ARN);
+ conf.put(AwsStsCredentialProvider.CONF_DURATION_SECONDS, "899");
+
+ AwsStsCredentialProvider provider = new AwsStsCredentialProvider();
+ IllegalArgumentException ex = assertThrows(IllegalArgumentException.class,
+ () -> provider.init(conf));
+
assertTrue(ex.getMessage().contains(AwsStsCredentialProvider.CONF_DURATION_SECONDS));
+ assertTrue(ex.getMessage().contains("900"));
+ assertTrue(ex.getMessage().contains("43200"));
+ }
+
+ @Test
+ public void testInitWithDurationAboveMaximumThrowsIllegalArgumentException()
{
+ Map<String, String> conf = new HashMap<>();
+ conf.put(AwsStsCredentialProvider.CONF_ROLE_ARN, TEST_ROLE_ARN);
+ conf.put(AwsStsCredentialProvider.CONF_DURATION_SECONDS, "43201");
+
+ AwsStsCredentialProvider provider = new AwsStsCredentialProvider();
+ IllegalArgumentException ex = assertThrows(IllegalArgumentException.class,
+ () -> provider.init(conf));
+
assertTrue(ex.getMessage().contains(AwsStsCredentialProvider.CONF_DURATION_SECONDS));
+ assertTrue(ex.getMessage().contains("900"));
+ assertTrue(ex.getMessage().contains("43200"));
+ }
+
+ @Test
+ public void testInitWithMinimumValidDurationSucceeds() {
+ Map<String, String> conf = new HashMap<>();
+ conf.put(AwsStsCredentialProvider.CONF_ROLE_ARN, TEST_ROLE_ARN);
+ conf.put(AwsStsCredentialProvider.CONF_DURATION_SECONDS, "900");
+
+ provider = new AwsStsCredentialProvider();
+ provider.init(conf);
+
+ assertNotNull(provider.resolvedConfig());
+ assertEquals(900, provider.resolvedConfig().durationSeconds);
+ }
+
+ @Test
+ public void testInitWithMaximumValidDurationSucceeds() {
+ Map<String, String> conf = new HashMap<>();
+ conf.put(AwsStsCredentialProvider.CONF_ROLE_ARN, TEST_ROLE_ARN);
+ conf.put(AwsStsCredentialProvider.CONF_DURATION_SECONDS, "43200");
+
+ provider = new AwsStsCredentialProvider();
+ provider.init(conf);
+
+ assertNotNull(provider.resolvedConfig());
+ assertEquals(43200, provider.resolvedConfig().durationSeconds);
+ }
+
+ @Test
+ public void testReInitializationThrowsIllegalStateException() {
+ Map<String, String> conf = new HashMap<>();
+ conf.put(AwsStsCredentialProvider.CONF_ROLE_ARN, TEST_ROLE_ARN);
+
+ provider = new AwsStsCredentialProvider();
+ provider.init(conf);
+
+ IllegalStateException ex = assertThrows(IllegalStateException.class,
+ () -> provider.init(conf));
+ assertTrue(ex.getMessage().contains("already initialized"));
+ }
+
+ @Test
+ public void testInitWithEndpointAndRegionResolvesCorrectly() {
+ Map<String, String> conf = new HashMap<>();
+ conf.put(AwsStsCredentialProvider.CONF_ROLE_ARN, TEST_ROLE_ARN);
+ conf.put(AwsStsCredentialProvider.CONF_STS_ENDPOINT,
"http://localhost:9000";);
+ conf.put(AwsStsCredentialProvider.CONF_REGION, "us-west-2");
+
+ provider = new AwsStsCredentialProvider();
+ provider.init(conf);
+
+ AwsStsCredentialProvider.ResolvedConfig cfg = provider.resolvedConfig();
+ assertNotNull(cfg);
+ assertEquals(Region.of("us-west-2"), cfg.resolvedRegion);
+ assertEquals(URI.create("http://localhost:9000";), cfg.endpointOverride);
+ assertNotNull(cfg.stsClient);
+ }
+
+ @Test
+ public void testInitWithEndpointNoRegionUsesDefault() {
+ Map<String, String> conf = new HashMap<>();
+ conf.put(AwsStsCredentialProvider.CONF_ROLE_ARN, TEST_ROLE_ARN);
+ conf.put(AwsStsCredentialProvider.CONF_STS_ENDPOINT,
"http://localhost:9000";);
+
+ provider = new AwsStsCredentialProvider();
+ provider.init(conf);
+
+ AwsStsCredentialProvider.ResolvedConfig cfg = provider.resolvedConfig();
+ assertNotNull(cfg);
+ assertEquals(Region.of("us-east-1"), cfg.resolvedRegion);
+ assertEquals(URI.create("http://localhost:9000";), cfg.endpointOverride);
+ }
+
+ @Test
+ public void testInitWithNeitherEndpointNorRegion() {
+ Map<String, String> conf = new HashMap<>();
+ conf.put(AwsStsCredentialProvider.CONF_ROLE_ARN, TEST_ROLE_ARN);
+
+ provider = new AwsStsCredentialProvider();
+ provider.init(conf);
+
+ AwsStsCredentialProvider.ResolvedConfig cfg = provider.resolvedConfig();
+ assertNotNull(cfg);
+ assertNull(cfg.resolvedRegion);
+ assertNull(cfg.endpointOverride);
}
@Test
- void testInitThrowsUnsupported() {
+ public void testInitWithMalformedEndpointThrowsIllegalArgumentException() {
+ Map<String, String> conf = new HashMap<>();
+ conf.put(AwsStsCredentialProvider.CONF_ROLE_ARN, TEST_ROLE_ARN);
+ conf.put(AwsStsCredentialProvider.CONF_STS_ENDPOINT, "not a valid uri^[]");
+
AwsStsCredentialProvider provider = new AwsStsCredentialProvider();
- assertThrows(UnsupportedOperationException.class, () ->
provider.init(Map.of()));
+ IllegalArgumentException ex = assertThrows(IllegalArgumentException.class,
+ () -> provider.init(conf));
+
assertTrue(ex.getMessage().contains(AwsStsCredentialProvider.CONF_STS_ENDPOINT),
+ "Error should mention the config key");
+ assertTrue(ex.getMessage().contains("not a valid uri^[]"),
+ "Error should mention the bad value");
+ assertNotNull(ex.getCause(), "Original IllegalArgumentException should be
preserved");
+ }
+
+ @Test
+ public void testInitTrimsConfigValues() {
+ Map<String, String> conf = new HashMap<>();
+ conf.put(AwsStsCredentialProvider.CONF_ROLE_ARN, " " + TEST_ROLE_ARN + "
");
+ conf.put(AwsStsCredentialProvider.CONF_STS_ENDPOINT, "
http://localhost:9000 ");
+ conf.put(AwsStsCredentialProvider.CONF_REGION, " us-west-2 ");
+
+ provider = new AwsStsCredentialProvider();
+ provider.init(conf);
+
+ AwsStsCredentialProvider.ResolvedConfig cfg = provider.resolvedConfig();
+ assertNotNull(cfg);
+ assertEquals(TEST_ROLE_ARN, cfg.roleArn);
+ assertEquals(Region.of("us-west-2"), cfg.resolvedRegion);
+ assertEquals(URI.create("http://localhost:9000";), cfg.endpointOverride);
+ }
+
+ @Test
+ public void testInitTrimsRoleSessionName() throws
CredentialResolutionException {
+ Instant expiration = Instant.now().plusSeconds(3600);
+ StsClient mockSts = createMockStsClient("AK", "SK", "ST", expiration);
+
+ // Use init() with a padded sessionName to exercise the trim path
+ Map<String, String> conf = new HashMap<>();
+ conf.put(AwsStsCredentialProvider.CONF_ROLE_ARN, TEST_ROLE_ARN);
+ conf.put(AwsStsCredentialProvider.CONF_SESSION_NAME, " my-session ");
+ conf.put(AwsStsCredentialProvider.CONF_STS_ENDPOINT,
"http://localhost:9000";);
+
+ provider = new AwsStsCredentialProvider();
+ provider.init(conf);
+
+ // Verify the stored config has the trimmed value
+ assertEquals("my-session", provider.resolvedConfig().roleSessionName);
+
+ // Now resolve() with a separate provider that uses the test constructor
+ // so we can capture the STS request via mock
+ String sessionName = provider.resolvedConfig().roleSessionName;
+ provider.close();
+ provider = new AwsStsCredentialProvider(
+ mockSts, TEST_ROLE_ARN, sessionName, null);
+
+ UserContext user = new UserContext(TEST_PRINCIPAL, TEST_ISSUER, TEST_TOKEN,
+ Instant.now(), Instant.now().plusSeconds(300));
+ provider.resolve(user, TEST_TARGET);
+
+ ArgumentCaptor<AssumeRoleWithWebIdentityRequest> captor =
+ ArgumentCaptor.forClass(AssumeRoleWithWebIdentityRequest.class);
+ verify(mockSts).assumeRoleWithWebIdentity(captor.capture());
+
+ // The session name in the request must be trimmed, not " my-session "
+ assertEquals("my-session", captor.getValue().roleSessionName());
+ }
+
+ // ========== resolve() ==========
+
+ @Test
+ public void testSuccessfulResolve() throws CredentialResolutionException {
+ Instant expiration = Instant.now().plusSeconds(3600);
+ StsClient mockSts = createMockStsClient("AKIAIOSFODNN7EXAMPLE",
+ "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY", "FwoGZXIvY...token",
expiration);
+
+ AwsStsCredentialProvider provider = new AwsStsCredentialProvider(
+ mockSts, TEST_ROLE_ARN, "test-session", 3600);
+
+ UserContext user = new UserContext(TEST_PRINCIPAL, TEST_ISSUER, TEST_TOKEN,
+ Instant.now(), Instant.now().plusSeconds(300));
+
+ ServiceCredential credential = provider.resolve(user, TEST_TARGET);
+
+ assertNotNull(credential);
+ assertEquals("AKIAIOSFODNN7EXAMPLE",
credential.getProperties().get("fs.s3a.access.key"));
+ assertEquals("wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
+ credential.getProperties().get("fs.s3a.secret.key"));
+ assertEquals("FwoGZXIvY...token",
credential.getProperties().get("fs.s3a.session.token"));
+ assertEquals(expiration, credential.getExpiresAt());
+ assertEquals(3, credential.getProperties().size());
+
+ // Verify the STS request was built correctly
+ ArgumentCaptor<AssumeRoleWithWebIdentityRequest> captor =
+ ArgumentCaptor.forClass(AssumeRoleWithWebIdentityRequest.class);
+ verify(mockSts).assumeRoleWithWebIdentity(captor.capture());
+
+ AssumeRoleWithWebIdentityRequest request = captor.getValue();
+ assertEquals(TEST_ROLE_ARN, request.roleArn());
+ assertEquals(TEST_TOKEN, request.webIdentityToken());
+ assertEquals("test-session", request.roleSessionName());
+ assertEquals(3600, request.durationSeconds());
+ }
+
+ @Test
+ public void testStsFailureWrappedInCredentialResolutionException() {
+ StsClient mockSts = mock(StsClient.class);
+ StsException stsException = (StsException) StsException.builder()
+ .message("Access denied for role")
+ .build();
+
when(mockSts.assumeRoleWithWebIdentity(any(AssumeRoleWithWebIdentityRequest.class)))
+ .thenThrow(stsException);
+
+ AwsStsCredentialProvider provider = new AwsStsCredentialProvider(
+ mockSts, TEST_ROLE_ARN, "test-session", null);
+
+ UserContext user = new UserContext(TEST_PRINCIPAL, TEST_ISSUER, TEST_TOKEN,
+ Instant.now(), Instant.now().plusSeconds(300));
+
+ CredentialResolutionException ex =
assertThrows(CredentialResolutionException.class,
+ () -> provider.resolve(user, TEST_TARGET));
+
+ // Verify the exception wraps the STS exception
+ assertEquals(stsException, ex.getCause());
+ assertTrue(ex.getMessage().contains(TEST_ROLE_ARN));
+ assertTrue(ex.getMessage().contains("AssumeRoleWithWebIdentity"));
+
+ // SECURITY: Verify that the raw token is NOT in any exception message
+ assertFalse(ex.getMessage().contains(TEST_TOKEN),
+ "Raw token must never appear in exception messages");
+ assertFalse(ex.getCause().getMessage() != null
+ && ex.getCause().getMessage().contains(TEST_TOKEN),
+ "Raw token must never appear in cause exception messages");
+ }
+
+ @Test
+ public void testTokenRedactedFromStsErrorMessage() {
+ StsClient mockSts = mock(StsClient.class);
+ // Simulate STS echoing the token back in an error message
+ StsException stsException = (StsException) StsException.builder()
+ .message("Invalid identity token: " + TEST_TOKEN)
+ .build();
+
when(mockSts.assumeRoleWithWebIdentity(any(AssumeRoleWithWebIdentityRequest.class)))
+ .thenThrow(stsException);
+
+ AwsStsCredentialProvider provider = new AwsStsCredentialProvider(
+ mockSts, TEST_ROLE_ARN, "test-session", null);
+
+ UserContext user = new UserContext(TEST_PRINCIPAL, TEST_ISSUER, TEST_TOKEN,
+ Instant.now(), Instant.now().plusSeconds(300));
+
+ CredentialResolutionException ex =
assertThrows(CredentialResolutionException.class,
+ () -> provider.resolve(user, TEST_TARGET));
+
+ // SECURITY: The wrapped message must NOT contain the raw token
+ assertFalse(ex.getMessage().contains(TEST_TOKEN),
+ "Raw token must be redacted from exception messages even if STS echoed
it");
+ assertTrue(ex.getMessage().contains("[REDACTED]"),
+ "Token should be replaced with [REDACTED]");
+
+ // SECURITY: The cause exception message must also NOT contain the raw
token
+ assertNotNull(ex.getCause(), "Cause should be a sanitized wrapper
exception");
+ assertFalse(ex.getCause().getMessage().contains(TEST_TOKEN),
+ "Raw token must be redacted from cause exception message");
+ }
+
+ @Test
+ public void testCustomSessionNameAndDurationPropagateToRequest()
+ throws CredentialResolutionException {
+ Instant expiration = Instant.now().plusSeconds(900);
+ StsClient mockSts = createMockStsClient("AK", "SK", "ST", expiration);
+
+ AwsStsCredentialProvider provider = new AwsStsCredentialProvider(
+ mockSts, TEST_ROLE_ARN, "custom-session-name", 900);
+
+ UserContext user = new UserContext(TEST_PRINCIPAL, TEST_ISSUER, TEST_TOKEN,
+ Instant.now(), Instant.now().plusSeconds(300));
+
+ provider.resolve(user, TEST_TARGET);
+
+ ArgumentCaptor<AssumeRoleWithWebIdentityRequest> captor =
+ ArgumentCaptor.forClass(AssumeRoleWithWebIdentityRequest.class);
+ verify(mockSts).assumeRoleWithWebIdentity(captor.capture());
+
+ AssumeRoleWithWebIdentityRequest request = captor.getValue();
+ assertEquals("custom-session-name", request.roleSessionName());
+ assertEquals(900, request.durationSeconds());
+ }
+
+ @Test
+ public void testNoDurationSecondsInRequestWhenNotConfigured()
+ throws CredentialResolutionException {
+ Instant expiration = Instant.now().plusSeconds(3600);
+ StsClient mockSts = createMockStsClient("AK", "SK", "ST", expiration);
+
+ AwsStsCredentialProvider provider = new AwsStsCredentialProvider(
+ mockSts, TEST_ROLE_ARN, "session", null);
+
+ UserContext user = new UserContext(TEST_PRINCIPAL, TEST_ISSUER, TEST_TOKEN,
+ Instant.now(), Instant.now().plusSeconds(300));
+
+ provider.resolve(user, TEST_TARGET);
+
+ ArgumentCaptor<AssumeRoleWithWebIdentityRequest> captor =
+ ArgumentCaptor.forClass(AssumeRoleWithWebIdentityRequest.class);
+ verify(mockSts).assumeRoleWithWebIdentity(captor.capture());
+
+ // durationSeconds should be null when not configured
+ assertNull(captor.getValue().durationSeconds());
+ }
+
+ @Test
+ public void testNullUserContextThrowsCredentialResolutionException() {
+ StsClient mockSts = mock(StsClient.class);
+ AwsStsCredentialProvider provider = new AwsStsCredentialProvider(
+ mockSts, TEST_ROLE_ARN, "session", null);
+
+ assertThrows(CredentialResolutionException.class,
+ () -> provider.resolve(null, TEST_TARGET));
+ }
+
+ @Test
+ public void testNullTargetThrowsCredentialResolutionException() {
+ StsClient mockSts = mock(StsClient.class);
+ AwsStsCredentialProvider provider = new AwsStsCredentialProvider(
+ mockSts, TEST_ROLE_ARN, "session", null);
+
+ UserContext user = new UserContext(TEST_PRINCIPAL, TEST_ISSUER, TEST_TOKEN,
+ Instant.now(), Instant.now().plusSeconds(300));
+
+ CredentialResolutionException ex =
assertThrows(CredentialResolutionException.class,
+ () -> provider.resolve(user, null));
+ assertTrue(ex.getMessage().contains("Target URI must not be null"));
+ }
+
+ @Test
+ public void testNullRawTokenThrowsCredentialResolutionException() {
+ StsClient mockSts = mock(StsClient.class);
+ AwsStsCredentialProvider provider = new AwsStsCredentialProvider(
+ mockSts, TEST_ROLE_ARN, "session", null);
+
+ UserContext user = mock(UserContext.class);
+ when(user.getRawToken()).thenReturn(null);
+
+ assertThrows(CredentialResolutionException.class,
+ () -> provider.resolve(user, TEST_TARGET));
+ }
+
+ @Test
+ public void testBlankRawTokenThrowsCredentialResolutionException() {
+ StsClient mockSts = mock(StsClient.class);
+ AwsStsCredentialProvider provider = new AwsStsCredentialProvider(
+ mockSts, TEST_ROLE_ARN, "session", null);
+
+ UserContext user = new UserContext(TEST_PRINCIPAL, TEST_ISSUER, " ",
+ Instant.now(), Instant.now().plusSeconds(300));
+
+ assertThrows(CredentialResolutionException.class,
+ () -> provider.resolve(user, TEST_TARGET));
}
@Test
- void testResolveThrowsUnsupported() {
+ public void testResolveBeforeInitThrowsCredentialResolutionException() {
AwsStsCredentialProvider provider = new AwsStsCredentialProvider();
- assertThrows(UnsupportedOperationException.class,
- () -> provider.resolve(null, URI.create("s3a://bucket/key")));
+
+ UserContext user = new UserContext(TEST_PRINCIPAL, TEST_ISSUER, TEST_TOKEN,
+ Instant.now(), Instant.now().plusSeconds(300));
+
+ CredentialResolutionException ex =
assertThrows(CredentialResolutionException.class,
+ () -> provider.resolve(user, TEST_TARGET));
+
+ assertTrue(ex.getMessage().contains("resolve() called before init()"));
}
@Test
- void testSuggestedTtl() {
+ public void testNullCredentialsResponseThrowsCredentialResolutionException()
{
+ StsClient mockSts = mock(StsClient.class);
+ AssumeRoleWithWebIdentityResponse response =
AssumeRoleWithWebIdentityResponse.builder()
+ .credentials((Credentials) null)
+ .build();
+
when(mockSts.assumeRoleWithWebIdentity(any(AssumeRoleWithWebIdentityRequest.class)))
+ .thenReturn(response);
+
+ AwsStsCredentialProvider provider = new AwsStsCredentialProvider(
+ mockSts, TEST_ROLE_ARN, "session", null);
+
+ UserContext user = new UserContext(TEST_PRINCIPAL, TEST_ISSUER, TEST_TOKEN,
+ Instant.now(), Instant.now().plusSeconds(300));
+
+ CredentialResolutionException ex =
assertThrows(CredentialResolutionException.class,
+ () -> provider.resolve(user, TEST_TARGET));
+
+ assertTrue(ex.getMessage().contains("incomplete credentials"));
+ assertTrue(ex.getMessage().contains(TEST_ROLE_ARN));
+ assertFalse(ex.getMessage().contains(TEST_TOKEN),
+ "Raw token must never appear in exception messages");
+ }
+
+ @Test
+ public void testMissingSessionTokenThrowsCredentialResolutionException() {
+ StsClient mockSts = mock(StsClient.class);
+ Credentials creds = Credentials.builder()
+ .accessKeyId("AKID")
+ .secretAccessKey("SECRET")
+ .sessionToken(null)
+ .expiration(Instant.now().plusSeconds(3600))
+ .build();
+ AssumeRoleWithWebIdentityResponse response =
AssumeRoleWithWebIdentityResponse.builder()
+ .credentials(creds)
+ .build();
+
when(mockSts.assumeRoleWithWebIdentity(any(AssumeRoleWithWebIdentityRequest.class)))
+ .thenReturn(response);
+
+ AwsStsCredentialProvider provider = new AwsStsCredentialProvider(
+ mockSts, TEST_ROLE_ARN, "session", null);
+
+ UserContext user = new UserContext(TEST_PRINCIPAL, TEST_ISSUER, TEST_TOKEN,
+ Instant.now(), Instant.now().plusSeconds(300));
+
+ CredentialResolutionException ex =
assertThrows(CredentialResolutionException.class,
+ () -> provider.resolve(user, TEST_TARGET));
+
+ assertTrue(ex.getMessage().contains("incomplete credentials"));
+ assertTrue(ex.getMessage().contains(TEST_ROLE_ARN));
+ assertFalse(ex.getMessage().contains(TEST_TOKEN),
+ "Raw token must never appear in exception messages");
+ }
+
+ // ========== supportedSchemes() ==========
+
+ @Test
+ public void testSupportedSchemesContainsS3a() {
AwsStsCredentialProvider provider = new AwsStsCredentialProvider();
+ Set<String> schemes = provider.supportedSchemes();
+ assertTrue(schemes.contains("s3a"));
+ assertEquals(1, schemes.size());
+ }
+
+ // ========== suggestedTtl() ==========
+
+ @Test
+ public void testSuggestedTtlWithDurationSeconds() {
+ StsClient mockSts = mock(StsClient.class);
+ AwsStsCredentialProvider provider = new AwsStsCredentialProvider(
+ mockSts, TEST_ROLE_ARN, "session", 1800);
+
+ assertEquals(Duration.ofSeconds(1800), provider.suggestedTtl());
+ }
+
+ @Test
+ public void testSuggestedTtlDefaultsTo15Minutes() {
+ StsClient mockSts = mock(StsClient.class);
+ AwsStsCredentialProvider provider = new AwsStsCredentialProvider(
+ mockSts, TEST_ROLE_ARN, "session", null);
+
assertEquals(Duration.ofMinutes(15), provider.suggestedTtl());
}
+
+ // ========== Session name derivation ==========
+
+ @Test
+ public void testDefaultSessionNameDerivedFromPrincipal()
+ throws CredentialResolutionException {
+ Instant expiration = Instant.now().plusSeconds(3600);
+ StsClient mockSts = createMockStsClient("AK", "SK", "ST", expiration);
+
+ AwsStsCredentialProvider provider = new AwsStsCredentialProvider(
+ mockSts, TEST_ROLE_ARN, null, null);
+
+ UserContext user = new UserContext("[email protected]", TEST_ISSUER,
TEST_TOKEN,
+ Instant.now(), Instant.now().plusSeconds(300));
+
+ provider.resolve(user, TEST_TARGET);
+
+ ArgumentCaptor<AssumeRoleWithWebIdentityRequest> captor =
+ ArgumentCaptor.forClass(AssumeRoleWithWebIdentityRequest.class);
+ verify(mockSts).assumeRoleWithWebIdentity(captor.capture());
+
+ String sessionName = captor.getValue().roleSessionName();
+ assertNotNull(sessionName);
+ assertFalse(sessionName.isBlank());
+ assertTrue(sessionName.contains("alice"));
+ // @ is valid in STS session names and should be preserved
+ assertTrue(sessionName.contains("@"));
+ }
+
+ @Test
+ public void testSanitizeSessionNameTruncatesLongPrincipal() {
+ // 70-char principal should be truncated to 64
+ String longPrincipal = "a".repeat(70);
+ String result =
AwsStsCredentialProvider.sanitizeSessionName(longPrincipal);
+ assertEquals(64, result.length());
+ assertEquals("a".repeat(64), result);
+ }
+
+ @Test
+ public void testSanitizeSessionNameReplacesInvalidChars() {
+ // Spaces, special chars should be replaced with '-'
+ String result = AwsStsCredentialProvider.sanitizeSessionName("user
name!#$%");
+ assertEquals("user-name----", result);
+ // Valid chars preserved
+ assertTrue(result.matches("[a-zA-Z0-9_+=,.@\\-]+"));
+ }
+
+ @Test
+ public void testSanitizeSessionNamePreservesValidChars() {
+ String validName = "user_+=,.@-test";
+ String result = AwsStsCredentialProvider.sanitizeSessionName(validName);
+ assertEquals(validName, result);
+ }
+
+ @Test
+ public void testSanitizeSessionNameFallsBackForShortResult() {
+ // Single char after sanitization -> falls back to default
+ String result = AwsStsCredentialProvider.sanitizeSessionName("x");
+ assertEquals("spark-oidc", result);
+ }
+
+ @Test
+ public void testSanitizeSessionNameFallsBackWhenAllInvalid() {
+ // All invalid chars replaced by '-', single char result
+ String result = AwsStsCredentialProvider.sanitizeSessionName("!");
+ assertEquals("spark-oidc", result);
+ }
+
+ @Test
+ public void testSanitizeSessionNameReplacesBackslash() {
+ // Regression: backslash must be replaced (invalid in STS session names)
+ String result =
AwsStsCredentialProvider.sanitizeSessionName("DOMAIN\\user");
+ assertEquals("DOMAIN-user", result);
+ }
+
+ // ========== close() ==========
+
+ @Test
+ public void testCloseShutsStsClient() {
+ StsClient mockSts = mock(StsClient.class);
+ AwsStsCredentialProvider provider = new AwsStsCredentialProvider(
+ mockSts, TEST_ROLE_ARN, "session", null);
+
+ provider.close();
+
+ verify(mockSts).close();
+ }
+
+ @Test
+ public void testCloseBeforeInitDoesNotThrow() {
+ AwsStsCredentialProvider provider = new AwsStsCredentialProvider();
+ // Should not throw even when config is null
+ provider.close();
+ }
+
+ // ========== Deep cause-chain token redaction (Item 1) ==========
+
+ @Test
+ public void testTokenInCauseChainIsRedacted() {
+ StsClient mockSts = mock(StsClient.class);
+ // Build a cause chain where the INNER cause contains the raw token,
+ // but the top-level message does NOT.
+ RuntimeException innerCause = new RuntimeException(
+ "Token validation failed: " + TEST_TOKEN);
+ StsException stsException = (StsException) StsException.builder()
+ .message("Access denied for role")
+ .cause(innerCause)
+ .build();
+
when(mockSts.assumeRoleWithWebIdentity(any(AssumeRoleWithWebIdentityRequest.class)))
+ .thenThrow(stsException);
+
+ AwsStsCredentialProvider p = new AwsStsCredentialProvider(
+ mockSts, TEST_ROLE_ARN, "test-session", null);
+
+ UserContext user = new UserContext(TEST_PRINCIPAL, TEST_ISSUER, TEST_TOKEN,
+ Instant.now(), Instant.now().plusSeconds(300));
+
+ CredentialResolutionException ex =
assertThrows(CredentialResolutionException.class,
+ () -> p.resolve(user, TEST_TARGET));
+
+ // Top-level message must not contain the raw token
+ assertFalse(ex.getMessage().contains(TEST_TOKEN),
+ "Raw token must not appear in top-level exception message");
+
+ // Walk the entire cause chain and assert the token is absent everywhere
+ Throwable current = ex.getCause();
+ while (current != null) {
+ if (current.getMessage() != null) {
+ assertFalse(current.getMessage().contains(TEST_TOKEN),
+ "Raw token must not appear in any cause message, but found in: "
+ + current.getClass().getSimpleName());
+ }
+ current = current.getCause();
+ }
+ }
+
+ // ========== Session name validation (Item 3) ==========
+
+ @Test
+ public void testInitWithValidCustomSessionName() {
+ Map<String, String> conf = new HashMap<>();
+ conf.put(AwsStsCredentialProvider.CONF_ROLE_ARN, TEST_ROLE_ARN);
+ conf.put(AwsStsCredentialProvider.CONF_SESSION_NAME,
"valid_session+=,.@-name");
+
+ provider = new AwsStsCredentialProvider();
+ provider.init(conf);
+
+ assertEquals("valid_session+=,.@-name",
provider.resolvedConfig().roleSessionName);
+ }
+
+ @Test
+ public void testInitWithInvalidSessionNameContainingSpace() {
+ Map<String, String> conf = new HashMap<>();
+ conf.put(AwsStsCredentialProvider.CONF_ROLE_ARN, TEST_ROLE_ARN);
+ conf.put(AwsStsCredentialProvider.CONF_SESSION_NAME, "bad session");
+
+ AwsStsCredentialProvider p = new AwsStsCredentialProvider();
+ IllegalArgumentException ex = assertThrows(IllegalArgumentException.class,
+ () -> p.init(conf));
+
assertTrue(ex.getMessage().contains(AwsStsCredentialProvider.CONF_SESSION_NAME),
+ "Error must name the config key");
+ assertTrue(ex.getMessage().contains("bad session"),
+ "Error must echo the bad value");
+ }
+
+ @Test
+ public void testInitWithInvalidSessionNameContainingQuestionMark() {
+ Map<String, String> conf = new HashMap<>();
+ conf.put(AwsStsCredentialProvider.CONF_ROLE_ARN, TEST_ROLE_ARN);
+ conf.put(AwsStsCredentialProvider.CONF_SESSION_NAME, "bad?name");
+
+ AwsStsCredentialProvider p = new AwsStsCredentialProvider();
+ IllegalArgumentException ex = assertThrows(IllegalArgumentException.class,
+ () -> p.init(conf));
+
assertTrue(ex.getMessage().contains(AwsStsCredentialProvider.CONF_SESSION_NAME));
+ }
+
+ @Test
+ public void testInitWithSessionNameTooShort() {
+ Map<String, String> conf = new HashMap<>();
+ conf.put(AwsStsCredentialProvider.CONF_ROLE_ARN, TEST_ROLE_ARN);
+ conf.put(AwsStsCredentialProvider.CONF_SESSION_NAME, "x");
+
+ AwsStsCredentialProvider p = new AwsStsCredentialProvider();
+ IllegalArgumentException ex = assertThrows(IllegalArgumentException.class,
+ () -> p.init(conf));
+
assertTrue(ex.getMessage().contains(AwsStsCredentialProvider.CONF_SESSION_NAME));
+ assertTrue(ex.getMessage().contains("x"));
+ }
+
+ @Test
+ public void testInitWithSessionNameTooLong() {
+ Map<String, String> conf = new HashMap<>();
+ conf.put(AwsStsCredentialProvider.CONF_ROLE_ARN, TEST_ROLE_ARN);
+ conf.put(AwsStsCredentialProvider.CONF_SESSION_NAME, "a".repeat(65));
+
+ AwsStsCredentialProvider p = new AwsStsCredentialProvider();
+ IllegalArgumentException ex = assertThrows(IllegalArgumentException.class,
+ () -> p.init(conf));
+
assertTrue(ex.getMessage().contains(AwsStsCredentialProvider.CONF_SESSION_NAME));
+ }
+
+ // ========== Session name: non-ASCII rejection (regression) ==========
+
+ @Test
+ public void testInitRejectsSessionNameWithAccentedChar() {
+ // "caf" + (char) 0xE9 produces "cafe" with accented 'e' -- valid under \w
but
+ // NOT valid in STS session names. This test would PASS (incorrectly)
under the
+ // buggy \w pattern and must FAIL (correctly) under the explicit ASCII
pattern.
+ // NOTE: (char) 0xNN construction avoids unicode escapes banned by Spark
+ // checkstyle (AvoidEscapedUnicodeCharacters) while keeping source bytes
ASCII.
+ Map<String, String> conf = new HashMap<>();
+ conf.put(AwsStsCredentialProvider.CONF_ROLE_ARN, TEST_ROLE_ARN);
+ conf.put(AwsStsCredentialProvider.CONF_SESSION_NAME, "caf" + (char) 0xE9);
+
+ AwsStsCredentialProvider p = new AwsStsCredentialProvider();
+ IllegalArgumentException ex = assertThrows(IllegalArgumentException.class,
+ () -> p.init(conf));
+
assertTrue(ex.getMessage().contains(AwsStsCredentialProvider.CONF_SESSION_NAME));
+ }
+
+ @Test
+ public void testInitRejectsSessionNameWithCjkChar() {
+ // CJK ideograph U+4E16 ('world' in Chinese) -- valid under \w but NOT
valid
+ // in STS session names. Regression test for the ASCII-only fix.
+ // NOTE: (char) 0xNN construction avoids unicode escapes banned by
checkstyle.
+ Map<String, String> conf = new HashMap<>();
+ conf.put(AwsStsCredentialProvider.CONF_ROLE_ARN, TEST_ROLE_ARN);
+ conf.put(AwsStsCredentialProvider.CONF_SESSION_NAME, "session" + (char)
0x4E16);
+
+ AwsStsCredentialProvider p = new AwsStsCredentialProvider();
+ IllegalArgumentException ex = assertThrows(IllegalArgumentException.class,
+ () -> p.init(conf));
+
assertTrue(ex.getMessage().contains(AwsStsCredentialProvider.CONF_SESSION_NAME));
+ }
+
+ // ========== close() idempotency ==========
+
+ @Test
+ public void testDoubleCloseIsIdempotent() {
+ StsClient mockSts = mock(StsClient.class);
+ AwsStsCredentialProvider p = new AwsStsCredentialProvider(
+ mockSts, TEST_ROLE_ARN, "test-session", null);
+
+ // First close should succeed normally
+ p.close();
+ // Second close must be a no-op (no exception, no double-close on the
client)
+ p.close();
+
+ // Verify the underlying client was closed exactly once
+ verify(mockSts).close();
+ }
+
+ @Test
+ public void testResolveAfterCloseThrowsWithClosedMessage() {
+ StsClient mockSts = mock(StsClient.class);
+ AwsStsCredentialProvider p = new AwsStsCredentialProvider(
+ mockSts, TEST_ROLE_ARN, "test-session", null);
+ p.close();
+
+ UserContext user = new UserContext(TEST_PRINCIPAL, TEST_ISSUER, TEST_TOKEN,
+ Instant.now(), Instant.now().plusSeconds(300));
+
+ CredentialResolutionException ex =
assertThrows(CredentialResolutionException.class,
+ () -> p.resolve(user, TEST_TARGET));
+ assertTrue(ex.getMessage().contains("closed"),
+ "Message should indicate the provider is closed");
+ }
+
+ // ========== causeChainStream cycle protection ==========
+
+ @Test
+ public void testCauseChainCycleDoesNotLoop() throws Exception {
+ // Construct a circular cause chain via reflection: nodeA -> nodeB ->
nodeA.
+ // Java's Throwable.initCause() prevents self-causation and
re-initialization,
+ // so we use Field.set to create the cycle.
+ RuntimeException nodeA = new RuntimeException("nodeA");
+ RuntimeException nodeB = new RuntimeException("nodeB", nodeA);
+ // nodeA.cause is currently the sentinel (this), which means "not yet set"
in
+ // OpenJDK's implementation. Force it to nodeB to create the cycle.
+ java.lang.reflect.Field causeField =
Throwable.class.getDeclaredField("cause");
+ causeField.setAccessible(true);
+ causeField.set(nodeA, nodeB);
+
+ // Now we have nodeA -> nodeB -> nodeA (a cycle).
+ // Simulate what resolve() does: call causeChainContainsToken and verify
termination.
+ StsClient mockSts = mock(StsClient.class);
+ StsException stsException = (StsException) StsException.builder()
+ .message("some error")
+ .cause(nodeA)
+ .build();
+
when(mockSts.assumeRoleWithWebIdentity(any(AssumeRoleWithWebIdentityRequest.class)))
+ .thenThrow(stsException);
+
+ AwsStsCredentialProvider p = new AwsStsCredentialProvider(
+ mockSts, TEST_ROLE_ARN, "test-session", null);
+
+ UserContext user = new UserContext(TEST_PRINCIPAL, TEST_ISSUER, TEST_TOKEN,
+ Instant.now(), Instant.now().plusSeconds(300));
+
+ // This must terminate (no infinite loop) and throw
CredentialResolutionException
+ assertThrows(CredentialResolutionException.class,
+ () -> p.resolve(user, TEST_TARGET));
+ }
+
+ // ========== resolve() wraps SDK IllegalStateException (Item 5) ==========
+
+ @Test
+ public void
testResolveWrapsSdkIllegalStateExceptionAsCredentialResolutionException() {
+ StsClient mockSts = mock(StsClient.class);
+
when(mockSts.assumeRoleWithWebIdentity(any(AssumeRoleWithWebIdentityRequest.class)))
+ .thenThrow(new IllegalStateException("client has been closed"));
+
+ AwsStsCredentialProvider p = new AwsStsCredentialProvider(
+ mockSts, TEST_ROLE_ARN, "test-session", null);
+ // Do NOT call p.close() -- keep the provider's closed flag false so that
+ // resolve() reaches the STS call, where the mock throws
IllegalStateException.
+ // This exercises the catch (IllegalStateException e) branch in resolve().
+
+ UserContext user = new UserContext(TEST_PRINCIPAL, TEST_ISSUER, TEST_TOKEN,
+ Instant.now(), Instant.now().plusSeconds(300));
+
+ CredentialResolutionException ex =
assertThrows(CredentialResolutionException.class,
+ () -> p.resolve(user, TEST_TARGET));
+
+ assertTrue(ex.getMessage().contains("closed"),
+ "Message should indicate the STS client has been closed");
+ assertFalse(ex.getMessage().contains(TEST_TOKEN),
+ "Raw token must never appear in exception messages");
+
+ // Verify the cause is the original IllegalStateException from the SDK
client
+ assertTrue(ex.getCause() instanceof IllegalStateException,
+ "Cause should be the IllegalStateException thrown by the STS client");
+ assertEquals("client has been closed", ex.getCause().getMessage());
+
+ // Verify the mock was actually invoked (proving the closed-flag
short-circuit
+ // was NOT hit and the test truly exercises the ISE catch branch)
+
verify(mockSts).assumeRoleWithWebIdentity(any(AssumeRoleWithWebIdentityRequest.class));
+ }
+
+ // ========== Helpers ==========
+
+ private StsClient createMockStsClient(String accessKeyId, String
secretAccessKey,
+ String sessionToken, Instant expiration) {
+ StsClient mockSts = mock(StsClient.class);
+ Credentials creds = Credentials.builder()
+ .accessKeyId(accessKeyId)
+ .secretAccessKey(secretAccessKey)
+ .sessionToken(sessionToken)
+ .expiration(expiration)
+ .build();
+ AssumeRoleWithWebIdentityResponse response =
AssumeRoleWithWebIdentityResponse.builder()
+ .credentials(creds)
+ .build();
+
when(mockSts.assumeRoleWithWebIdentity(any(AssumeRoleWithWebIdentityRequest.class)))
+ .thenReturn(response);
+ return mockSts;
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
