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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
