sarutak commented on code in PR #57655: URL: https://github.com/apache/spark/pull/57655#discussion_r3687938661
########## connector/credential-aws/src/main/java/org/apache/spark/security/aws/AwsStsCredentialProvider.java: ########## @@ -0,0 +1,316 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.security.aws; + +import java.net.URI; +import java.time.Duration; +import java.time.Instant; +import java.util.Map; +import java.util.Set; + +import software.amazon.awssdk.auth.credentials.AnonymousCredentialsProvider; +import software.amazon.awssdk.core.exception.SdkException; +import software.amazon.awssdk.regions.Region; +import software.amazon.awssdk.services.sts.StsClient; +import software.amazon.awssdk.services.sts.StsClientBuilder; +import software.amazon.awssdk.services.sts.model.AssumeRoleWithWebIdentityRequest; +import software.amazon.awssdk.services.sts.model.AssumeRoleWithWebIdentityResponse; +import software.amazon.awssdk.services.sts.model.Credentials; + +import org.apache.spark.annotation.DeveloperApi; +import org.apache.spark.security.CredentialProvider; +import org.apache.spark.security.CredentialResolutionException; +import org.apache.spark.security.ServiceCredential; +import org.apache.spark.security.UserContext; + +/** + * :: DeveloperApi :: + * A {@link CredentialProvider} that exchanges an OIDC identity token for temporary + * AWS credentials via the STS {@code AssumeRoleWithWebIdentity} API. + * <p> + * The returned {@link ServiceCredential} contains S3A-compatible Hadoop configuration + * properties ({@code fs.s3a.access.key}, {@code fs.s3a.secret.key}, + * {@code fs.s3a.session.token}) that can be propagated to executors for accessing + * S3-compatible storage. + * <p> + * <b>Configuration keys</b> (passed via {@code spark.security.oidc.*}): + * <ul> + * <li>{@code spark.security.oidc.roleArn} (required) — the ARN of the IAM role to assume</li> + * <li>{@code spark.security.oidc.sessionName} (optional) — the role session name; + * defaults to a value derived from the user's principal or "spark-oidc"</li> + * <li>{@code spark.security.oidc.durationSeconds} (optional) — credential duration in seconds; + * if unset, STS uses the role's default maximum</li> + * <li>{@code spark.security.oidc.region} (optional) — the AWS region for the STS endpoint; + * defaults to us-east-1 when only {@code stsEndpoint} is set</li> + * <li>{@code spark.security.oidc.stsEndpoint} (optional) — a custom STS endpoint URL for + * non-AWS environments (MinIO, Ceph, LocalStack, etc.)</li> + * </ul> + * <p> + * <b>Security note:</b> The OIDC raw token is never included in log messages or + * exception messages. It is passed directly to the STS API call and discarded. + * + * @since 4.3.0 + */ +@DeveloperApi +public class AwsStsCredentialProvider implements CredentialProvider { + + // Configuration key constants + static final String CONF_ROLE_ARN = "spark.security.oidc.roleArn"; Review Comment: Could you rename to `spark.security.oidc.aws.*` to align with the SPIP doc? ########## pom.xml: ########## @@ -3519,6 +3519,13 @@ </modules> </profile> + <profile> + <id>credential-aws</id> + <modules> + <module>connector/credential-aws</module> Review Comment: Since introducing a new Maven profile requires updates to many CI/release/lint scripts, I'd like to split the module scaffolding (SPARK-57897) into a separate PR as originally anticipated. I'll handle it. It will include: - Module definition (`connector/credential-aws/pom.xml` with a stub provider) - `pom.xml` profile, `SparkBuild.scala`, `modules.py`, `utils.py` - `META-INF/services` registration - CI/release/lint script updates: - `.github/workflows/build_and_test.yml` - `.github/workflows/maven_test.yml` - `.github/workflows/python_hosted_runner_test.yml` - `.github/workflows/benchmark.yml` - `dev/create-release/release-build.sh` - `dev/test-dependencies.sh` - `dev/mima` - `dev/lint-java` - `dev/sbt-checkstyle` - `dev/scalastyle` - `docs/_plugins/build_api_docs.rb` - `dev/spark-test-image-util/docs/build-docs` Once that's merged, this PR can focus purely on the `AwsStsCredentialProvider` implementation and tests, removing the build wiring changes from its diff. ########## connector/credential-aws/src/main/java/org/apache/spark/security/aws/AwsStsCredentialProvider.java: ########## @@ -0,0 +1,316 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.security.aws; + +import java.net.URI; +import java.time.Duration; +import java.time.Instant; +import java.util.Map; +import java.util.Set; + +import software.amazon.awssdk.auth.credentials.AnonymousCredentialsProvider; +import software.amazon.awssdk.core.exception.SdkException; +import software.amazon.awssdk.regions.Region; +import software.amazon.awssdk.services.sts.StsClient; +import software.amazon.awssdk.services.sts.StsClientBuilder; +import software.amazon.awssdk.services.sts.model.AssumeRoleWithWebIdentityRequest; +import software.amazon.awssdk.services.sts.model.AssumeRoleWithWebIdentityResponse; +import software.amazon.awssdk.services.sts.model.Credentials; + +import org.apache.spark.annotation.DeveloperApi; +import org.apache.spark.security.CredentialProvider; +import org.apache.spark.security.CredentialResolutionException; +import org.apache.spark.security.ServiceCredential; +import org.apache.spark.security.UserContext; + +/** + * :: DeveloperApi :: + * A {@link CredentialProvider} that exchanges an OIDC identity token for temporary + * AWS credentials via the STS {@code AssumeRoleWithWebIdentity} API. + * <p> + * The returned {@link ServiceCredential} contains S3A-compatible Hadoop configuration + * properties ({@code fs.s3a.access.key}, {@code fs.s3a.secret.key}, + * {@code fs.s3a.session.token}) that can be propagated to executors for accessing + * S3-compatible storage. + * <p> + * <b>Configuration keys</b> (passed via {@code spark.security.oidc.*}): + * <ul> + * <li>{@code spark.security.oidc.roleArn} (required) — the ARN of the IAM role to assume</li> + * <li>{@code spark.security.oidc.sessionName} (optional) — the role session name; + * defaults to a value derived from the user's principal or "spark-oidc"</li> + * <li>{@code spark.security.oidc.durationSeconds} (optional) — credential duration in seconds; + * if unset, STS uses the role's default maximum</li> + * <li>{@code spark.security.oidc.region} (optional) — the AWS region for the STS endpoint; + * defaults to us-east-1 when only {@code stsEndpoint} is set</li> + * <li>{@code spark.security.oidc.stsEndpoint} (optional) — a custom STS endpoint URL for + * non-AWS environments (MinIO, Ceph, LocalStack, etc.)</li> + * </ul> + * <p> + * <b>Security note:</b> The OIDC raw token is never included in log messages or + * exception messages. It is passed directly to the STS API call and discarded. + * + * @since 4.3.0 + */ +@DeveloperApi +public class AwsStsCredentialProvider implements CredentialProvider { Review Comment: `StsClient` implements `SdkAutoCloseable` and holds HTTP connection pools internally. The current implementation creates the client in `init()` but never closes it. `CredentialProvider` uses an `init()` pattern where implementations construct long-lived resources. This means the SPI should provide a lifecycle hook for cleanup. I'll open a PR to add `AutoCloseable` to `CredentialProvider` (with a default no-op `close()`) and wire `UserCredentialManager.stop()` to call `provider.close()`. Once that's merged, could you implement `close()` here to shut down the `StsClient`? In the meantime, could you also add a re-initialization guard at the top of `init()`? ```java if (this.config != null) { throw new IllegalStateException("AwsStsCredentialProvider is already initialized"); } ``` ########## connector/credential-aws/src/main/java/org/apache/spark/security/aws/AwsStsCredentialProvider.java: ########## @@ -0,0 +1,316 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.security.aws; + +import java.net.URI; +import java.time.Duration; +import java.time.Instant; +import java.util.Map; +import java.util.Set; + +import software.amazon.awssdk.auth.credentials.AnonymousCredentialsProvider; +import software.amazon.awssdk.core.exception.SdkException; +import software.amazon.awssdk.regions.Region; +import software.amazon.awssdk.services.sts.StsClient; +import software.amazon.awssdk.services.sts.StsClientBuilder; +import software.amazon.awssdk.services.sts.model.AssumeRoleWithWebIdentityRequest; +import software.amazon.awssdk.services.sts.model.AssumeRoleWithWebIdentityResponse; +import software.amazon.awssdk.services.sts.model.Credentials; + +import org.apache.spark.annotation.DeveloperApi; +import org.apache.spark.security.CredentialProvider; +import org.apache.spark.security.CredentialResolutionException; +import org.apache.spark.security.ServiceCredential; +import org.apache.spark.security.UserContext; + +/** + * :: DeveloperApi :: + * A {@link CredentialProvider} that exchanges an OIDC identity token for temporary + * AWS credentials via the STS {@code AssumeRoleWithWebIdentity} API. + * <p> + * The returned {@link ServiceCredential} contains S3A-compatible Hadoop configuration + * properties ({@code fs.s3a.access.key}, {@code fs.s3a.secret.key}, + * {@code fs.s3a.session.token}) that can be propagated to executors for accessing + * S3-compatible storage. + * <p> + * <b>Configuration keys</b> (passed via {@code spark.security.oidc.*}): + * <ul> + * <li>{@code spark.security.oidc.roleArn} (required) — the ARN of the IAM role to assume</li> + * <li>{@code spark.security.oidc.sessionName} (optional) — the role session name; + * defaults to a value derived from the user's principal or "spark-oidc"</li> + * <li>{@code spark.security.oidc.durationSeconds} (optional) — credential duration in seconds; + * if unset, STS uses the role's default maximum</li> + * <li>{@code spark.security.oidc.region} (optional) — the AWS region for the STS endpoint; + * defaults to us-east-1 when only {@code stsEndpoint} is set</li> + * <li>{@code spark.security.oidc.stsEndpoint} (optional) — a custom STS endpoint URL for + * non-AWS environments (MinIO, Ceph, LocalStack, etc.)</li> + * </ul> + * <p> + * <b>Security note:</b> The OIDC raw token is never included in log messages or + * exception messages. It is passed directly to the STS API call and discarded. + * + * @since 4.3.0 + */ +@DeveloperApi +public class AwsStsCredentialProvider implements CredentialProvider { + + // Configuration key constants + static final String CONF_ROLE_ARN = "spark.security.oidc.roleArn"; + static final String CONF_SESSION_NAME = "spark.security.oidc.sessionName"; + static final String CONF_DURATION_SECONDS = "spark.security.oidc.durationSeconds"; + static final String CONF_REGION = "spark.security.oidc.region"; + static final String CONF_STS_ENDPOINT = "spark.security.oidc.stsEndpoint"; + + private static final String DEFAULT_REGION = "us-east-1"; + private static final String DEFAULT_SESSION_NAME = "spark-oidc"; + + /** + * Immutable configuration holder that is safely published via the volatile + * {@link #config} field. All fields are set once during construction and are + * final, ensuring correct visibility across threads after init() completes. + */ + static final class ResolvedConfig { + final String roleArn; + final String roleSessionName; + final Integer durationSeconds; + final Region resolvedRegion; + final URI endpointOverride; + final StsClient stsClient; + + ResolvedConfig(String roleArn, String roleSessionName, Integer durationSeconds, + Region resolvedRegion, URI endpointOverride, StsClient stsClient) { + this.roleArn = roleArn; + this.roleSessionName = roleSessionName; + this.durationSeconds = durationSeconds; + this.resolvedRegion = resolvedRegion; + this.endpointOverride = endpointOverride; + this.stsClient = stsClient; + } + } + + /** Safely published via volatile write in init(); read in resolve()/suggestedTtl(). */ + private volatile ResolvedConfig config; + + /** + * Default no-arg constructor used by {@link java.util.ServiceLoader}. + */ + public AwsStsCredentialProvider() { + // ServiceLoader requires a public no-arg constructor + } + + /** + * Package-private constructor for testing with an injected STS client. + * + * @param stsClient the STS client to use (must not be null) + * @param roleArn the IAM role ARN (must not be null or blank) + * @param roleSessionName the session name (may be null for default) + * @param durationSeconds the credential duration in seconds (may be null) + */ + AwsStsCredentialProvider(StsClient stsClient, String roleArn, String roleSessionName, Review Comment: Please add `@VisibleForTesting` to this package-private constructor. ########## connector/credential-aws/src/main/java/org/apache/spark/security/aws/AwsStsCredentialProvider.java: ########## @@ -0,0 +1,316 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.security.aws; + +import java.net.URI; +import java.time.Duration; +import java.time.Instant; +import java.util.Map; +import java.util.Set; + +import software.amazon.awssdk.auth.credentials.AnonymousCredentialsProvider; +import software.amazon.awssdk.core.exception.SdkException; +import software.amazon.awssdk.regions.Region; +import software.amazon.awssdk.services.sts.StsClient; +import software.amazon.awssdk.services.sts.StsClientBuilder; +import software.amazon.awssdk.services.sts.model.AssumeRoleWithWebIdentityRequest; +import software.amazon.awssdk.services.sts.model.AssumeRoleWithWebIdentityResponse; +import software.amazon.awssdk.services.sts.model.Credentials; + +import org.apache.spark.annotation.DeveloperApi; +import org.apache.spark.security.CredentialProvider; +import org.apache.spark.security.CredentialResolutionException; +import org.apache.spark.security.ServiceCredential; +import org.apache.spark.security.UserContext; + +/** + * :: DeveloperApi :: + * A {@link CredentialProvider} that exchanges an OIDC identity token for temporary + * AWS credentials via the STS {@code AssumeRoleWithWebIdentity} API. + * <p> + * The returned {@link ServiceCredential} contains S3A-compatible Hadoop configuration + * properties ({@code fs.s3a.access.key}, {@code fs.s3a.secret.key}, + * {@code fs.s3a.session.token}) that can be propagated to executors for accessing + * S3-compatible storage. + * <p> + * <b>Configuration keys</b> (passed via {@code spark.security.oidc.*}): + * <ul> + * <li>{@code spark.security.oidc.roleArn} (required) — the ARN of the IAM role to assume</li> + * <li>{@code spark.security.oidc.sessionName} (optional) — the role session name; + * defaults to a value derived from the user's principal or "spark-oidc"</li> + * <li>{@code spark.security.oidc.durationSeconds} (optional) — credential duration in seconds; + * if unset, STS uses the role's default maximum</li> + * <li>{@code spark.security.oidc.region} (optional) — the AWS region for the STS endpoint; + * defaults to us-east-1 when only {@code stsEndpoint} is set</li> + * <li>{@code spark.security.oidc.stsEndpoint} (optional) — a custom STS endpoint URL for + * non-AWS environments (MinIO, Ceph, LocalStack, etc.)</li> + * </ul> + * <p> + * <b>Security note:</b> The OIDC raw token is never included in log messages or + * exception messages. It is passed directly to the STS API call and discarded. + * + * @since 4.3.0 + */ +@DeveloperApi +public class AwsStsCredentialProvider implements CredentialProvider { + + // Configuration key constants + static final String CONF_ROLE_ARN = "spark.security.oidc.roleArn"; + static final String CONF_SESSION_NAME = "spark.security.oidc.sessionName"; + static final String CONF_DURATION_SECONDS = "spark.security.oidc.durationSeconds"; + static final String CONF_REGION = "spark.security.oidc.region"; + static final String CONF_STS_ENDPOINT = "spark.security.oidc.stsEndpoint"; + + private static final String DEFAULT_REGION = "us-east-1"; + private static final String DEFAULT_SESSION_NAME = "spark-oidc"; + + /** + * Immutable configuration holder that is safely published via the volatile + * {@link #config} field. All fields are set once during construction and are + * final, ensuring correct visibility across threads after init() completes. + */ + static final class ResolvedConfig { + final String roleArn; + final String roleSessionName; + final Integer durationSeconds; + final Region resolvedRegion; + final URI endpointOverride; + final StsClient stsClient; + + ResolvedConfig(String roleArn, String roleSessionName, Integer durationSeconds, + Region resolvedRegion, URI endpointOverride, StsClient stsClient) { + this.roleArn = roleArn; + this.roleSessionName = roleSessionName; + this.durationSeconds = durationSeconds; + this.resolvedRegion = resolvedRegion; + this.endpointOverride = endpointOverride; + this.stsClient = stsClient; + } + } + + /** Safely published via volatile write in init(); read in resolve()/suggestedTtl(). */ + private volatile ResolvedConfig config; + + /** + * Default no-arg constructor used by {@link java.util.ServiceLoader}. + */ + public AwsStsCredentialProvider() { + // ServiceLoader requires a public no-arg constructor + } + + /** + * Package-private constructor for testing with an injected STS client. + * + * @param stsClient the STS client to use (must not be null) + * @param roleArn the IAM role ARN (must not be null or blank) + * @param roleSessionName the session name (may be null for default) + * @param durationSeconds the credential duration in seconds (may be null) + */ + AwsStsCredentialProvider(StsClient stsClient, String roleArn, String roleSessionName, + Integer durationSeconds) { + this.config = new ResolvedConfig(roleArn, roleSessionName, durationSeconds, + null, null, stsClient); + } + + @Override + public void init(Map<String, String> conf) { + String roleArn = conf.get(CONF_ROLE_ARN); + if (roleArn == null || roleArn.isBlank()) { + throw new IllegalArgumentException( + "Configuration key '" + CONF_ROLE_ARN + "' is required but was not set. " + + "Specify the ARN of the IAM role to assume via AssumeRoleWithWebIdentity."); + } + + String roleSessionName = conf.get(CONF_SESSION_NAME); + String regionStr = conf.get(CONF_REGION); + String stsEndpoint = conf.get(CONF_STS_ENDPOINT); + + Integer durationSeconds = null; + String durationStr = conf.get(CONF_DURATION_SECONDS); + if (durationStr != null && !durationStr.isBlank()) { + try { + durationSeconds = Integer.parseInt(durationStr.trim()); + if (durationSeconds <= 0) { + throw new IllegalArgumentException( + "Configuration key '" + CONF_DURATION_SECONDS + + "' must be a positive integer, got: " + durationStr); + } + } catch (NumberFormatException e) { + throw new IllegalArgumentException( + "Configuration key '" + CONF_DURATION_SECONDS + + "' must be a valid integer, got: " + durationStr, e); + } + } + + // Resolve the region and endpoint before building the client + Region resolvedRegion = resolveRegion(regionStr, stsEndpoint); + URI endpointOverride = resolveEndpoint(stsEndpoint); + + StsClient stsClient = buildStsClient(resolvedRegion, endpointOverride); + + // Single volatile write publishes all configuration atomically + this.config = new ResolvedConfig(roleArn, roleSessionName, durationSeconds, + resolvedRegion, endpointOverride, stsClient); + } + + /** + * Resolves the AWS region based on explicit configuration and endpoint presence. + * When a custom endpoint is provided without an explicit region, defaults to us-east-1. + */ + private static Region resolveRegion(String regionStr, String stsEndpoint) { + if (regionStr != null && !regionStr.isBlank()) { + return Region.of(regionStr); + } else if (stsEndpoint != null && !stsEndpoint.isBlank()) { + // When a custom endpoint is set but no explicit region, use a default region. + // The region is required by the SDK but not meaningful for non-AWS endpoints. + return Region.of(DEFAULT_REGION); + } + return null; + } + + /** + * Resolves the endpoint override URI from configuration. + */ + private static URI resolveEndpoint(String stsEndpoint) { + if (stsEndpoint != null && !stsEndpoint.isBlank()) { + return URI.create(stsEndpoint); + } + return null; + } + + /** + * Builds the STS client with the resolved region and endpoint. + */ + private static StsClient buildStsClient(Region resolvedRegion, URI endpointOverride) { + StsClientBuilder builder = StsClient.builder() + .credentialsProvider(AnonymousCredentialsProvider.create()); Review Comment: A brief comment explaining why anonymous credentials are used would help future readers: ```java // AssumeRoleWithWebIdentity does not require AWS credentials; // the OIDC token itself serves as the authentication mechanism. .credentialsProvider(AnonymousCredentialsProvider.create()) ``` ########## connector/credential-aws/src/main/java/org/apache/spark/security/aws/AwsStsCredentialProvider.java: ########## @@ -0,0 +1,316 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.security.aws; + +import java.net.URI; +import java.time.Duration; +import java.time.Instant; +import java.util.Map; +import java.util.Set; + +import software.amazon.awssdk.auth.credentials.AnonymousCredentialsProvider; +import software.amazon.awssdk.core.exception.SdkException; +import software.amazon.awssdk.regions.Region; +import software.amazon.awssdk.services.sts.StsClient; +import software.amazon.awssdk.services.sts.StsClientBuilder; +import software.amazon.awssdk.services.sts.model.AssumeRoleWithWebIdentityRequest; +import software.amazon.awssdk.services.sts.model.AssumeRoleWithWebIdentityResponse; +import software.amazon.awssdk.services.sts.model.Credentials; + +import org.apache.spark.annotation.DeveloperApi; +import org.apache.spark.security.CredentialProvider; +import org.apache.spark.security.CredentialResolutionException; +import org.apache.spark.security.ServiceCredential; +import org.apache.spark.security.UserContext; + +/** + * :: DeveloperApi :: + * A {@link CredentialProvider} that exchanges an OIDC identity token for temporary + * AWS credentials via the STS {@code AssumeRoleWithWebIdentity} API. + * <p> + * The returned {@link ServiceCredential} contains S3A-compatible Hadoop configuration + * properties ({@code fs.s3a.access.key}, {@code fs.s3a.secret.key}, + * {@code fs.s3a.session.token}) that can be propagated to executors for accessing + * S3-compatible storage. + * <p> + * <b>Configuration keys</b> (passed via {@code spark.security.oidc.*}): + * <ul> + * <li>{@code spark.security.oidc.roleArn} (required) — the ARN of the IAM role to assume</li> + * <li>{@code spark.security.oidc.sessionName} (optional) — the role session name; + * defaults to a value derived from the user's principal or "spark-oidc"</li> + * <li>{@code spark.security.oidc.durationSeconds} (optional) — credential duration in seconds; + * if unset, STS uses the role's default maximum</li> + * <li>{@code spark.security.oidc.region} (optional) — the AWS region for the STS endpoint; + * defaults to us-east-1 when only {@code stsEndpoint} is set</li> + * <li>{@code spark.security.oidc.stsEndpoint} (optional) — a custom STS endpoint URL for + * non-AWS environments (MinIO, Ceph, LocalStack, etc.)</li> + * </ul> + * <p> + * <b>Security note:</b> The OIDC raw token is never included in log messages or + * exception messages. It is passed directly to the STS API call and discarded. + * + * @since 4.3.0 + */ +@DeveloperApi +public class AwsStsCredentialProvider implements CredentialProvider { + + // Configuration key constants + static final String CONF_ROLE_ARN = "spark.security.oidc.roleArn"; + static final String CONF_SESSION_NAME = "spark.security.oidc.sessionName"; + static final String CONF_DURATION_SECONDS = "spark.security.oidc.durationSeconds"; + static final String CONF_REGION = "spark.security.oidc.region"; + static final String CONF_STS_ENDPOINT = "spark.security.oidc.stsEndpoint"; + + private static final String DEFAULT_REGION = "us-east-1"; + private static final String DEFAULT_SESSION_NAME = "spark-oidc"; + + /** + * Immutable configuration holder that is safely published via the volatile + * {@link #config} field. All fields are set once during construction and are + * final, ensuring correct visibility across threads after init() completes. + */ + static final class ResolvedConfig { + final String roleArn; + final String roleSessionName; + final Integer durationSeconds; + final Region resolvedRegion; + final URI endpointOverride; + final StsClient stsClient; + + ResolvedConfig(String roleArn, String roleSessionName, Integer durationSeconds, + Region resolvedRegion, URI endpointOverride, StsClient stsClient) { + this.roleArn = roleArn; + this.roleSessionName = roleSessionName; + this.durationSeconds = durationSeconds; + this.resolvedRegion = resolvedRegion; + this.endpointOverride = endpointOverride; + this.stsClient = stsClient; + } + } + + /** Safely published via volatile write in init(); read in resolve()/suggestedTtl(). */ + private volatile ResolvedConfig config; + + /** + * Default no-arg constructor used by {@link java.util.ServiceLoader}. + */ + public AwsStsCredentialProvider() { + // ServiceLoader requires a public no-arg constructor + } + + /** + * Package-private constructor for testing with an injected STS client. + * + * @param stsClient the STS client to use (must not be null) + * @param roleArn the IAM role ARN (must not be null or blank) + * @param roleSessionName the session name (may be null for default) + * @param durationSeconds the credential duration in seconds (may be null) + */ + AwsStsCredentialProvider(StsClient stsClient, String roleArn, String roleSessionName, + Integer durationSeconds) { + this.config = new ResolvedConfig(roleArn, roleSessionName, durationSeconds, + null, null, stsClient); + } + + @Override + public void init(Map<String, String> conf) { + String roleArn = conf.get(CONF_ROLE_ARN); + if (roleArn == null || roleArn.isBlank()) { + throw new IllegalArgumentException( + "Configuration key '" + CONF_ROLE_ARN + "' is required but was not set. " + + "Specify the ARN of the IAM role to assume via AssumeRoleWithWebIdentity."); + } + + String roleSessionName = conf.get(CONF_SESSION_NAME); + String regionStr = conf.get(CONF_REGION); + String stsEndpoint = conf.get(CONF_STS_ENDPOINT); + + Integer durationSeconds = null; + String durationStr = conf.get(CONF_DURATION_SECONDS); + if (durationStr != null && !durationStr.isBlank()) { + try { + durationSeconds = Integer.parseInt(durationStr.trim()); + if (durationSeconds <= 0) { + throw new IllegalArgumentException( + "Configuration key '" + CONF_DURATION_SECONDS + + "' must be a positive integer, got: " + durationStr); + } + } catch (NumberFormatException e) { + throw new IllegalArgumentException( + "Configuration key '" + CONF_DURATION_SECONDS + + "' must be a valid integer, got: " + durationStr, e); + } + } + + // Resolve the region and endpoint before building the client + Region resolvedRegion = resolveRegion(regionStr, stsEndpoint); + URI endpointOverride = resolveEndpoint(stsEndpoint); + + StsClient stsClient = buildStsClient(resolvedRegion, endpointOverride); + + // Single volatile write publishes all configuration atomically + this.config = new ResolvedConfig(roleArn, roleSessionName, durationSeconds, + resolvedRegion, endpointOverride, stsClient); + } + + /** + * Resolves the AWS region based on explicit configuration and endpoint presence. + * When a custom endpoint is provided without an explicit region, defaults to us-east-1. + */ + private static Region resolveRegion(String regionStr, String stsEndpoint) { + if (regionStr != null && !regionStr.isBlank()) { + return Region.of(regionStr); + } else if (stsEndpoint != null && !stsEndpoint.isBlank()) { + // When a custom endpoint is set but no explicit region, use a default region. + // The region is required by the SDK but not meaningful for non-AWS endpoints. + return Region.of(DEFAULT_REGION); + } + return null; + } + + /** + * Resolves the endpoint override URI from configuration. + */ + private static URI resolveEndpoint(String stsEndpoint) { + if (stsEndpoint != null && !stsEndpoint.isBlank()) { + return URI.create(stsEndpoint); + } + return null; + } + + /** + * Builds the STS client with the resolved region and endpoint. + */ + private static StsClient buildStsClient(Region resolvedRegion, URI endpointOverride) { + StsClientBuilder builder = StsClient.builder() + .credentialsProvider(AnonymousCredentialsProvider.create()); + + if (resolvedRegion != null) { + builder.region(resolvedRegion); + } + + if (endpointOverride != null) { + builder.endpointOverride(endpointOverride); + } + + return builder.build(); + } + + /** + * Returns the resolved configuration for testing purposes. + * Package-private visibility allows test assertions on resolved region/endpoint. + */ + ResolvedConfig resolvedConfig() { + return config; + } + + @Override + public Set<String> supportedSchemes() { + return Set.of("s3a"); + } + + @Override + public ServiceCredential resolve(UserContext user, URI target) + throws CredentialResolutionException { + if (user == null) { + throw new CredentialResolutionException( + "UserContext must not be null when resolving AWS credentials"); + } + if (user.getRawToken() == null || user.getRawToken().isBlank()) { + throw new CredentialResolutionException( + "UserContext raw token must not be null or blank; cannot perform " + + "AssumeRoleWithWebIdentity without an identity token"); + } + + ResolvedConfig cfg = this.config; + if (cfg == null) { + throw new CredentialResolutionException("resolve() called before init()"); + } + String sessionName = cfg.roleSessionName; + if (sessionName == null || sessionName.isBlank()) { + // Derive from principal, sanitizing for STS session name constraints + // (alphanumeric, =,.@- only, max 64 chars) + String principal = user.getPrincipal(); + if (principal != null && !principal.isBlank()) { + sessionName = sanitizeSessionName(principal); + } else { + sessionName = DEFAULT_SESSION_NAME; + } + } + + try { + AssumeRoleWithWebIdentityRequest.Builder reqBuilder = + AssumeRoleWithWebIdentityRequest.builder() + .roleArn(cfg.roleArn) + .roleSessionName(sessionName) + .webIdentityToken(user.getRawToken()); + + if (cfg.durationSeconds != null) { + reqBuilder.durationSeconds(cfg.durationSeconds); + } + + AssumeRoleWithWebIdentityResponse response = + cfg.stsClient.assumeRoleWithWebIdentity(reqBuilder.build()); + + Credentials creds = response.credentials(); + if (creds == null || creds.accessKeyId() == null + || creds.secretAccessKey() == null || creds.sessionToken() == null) { + throw new CredentialResolutionException( + "STS returned incomplete credentials for role '" + cfg.roleArn + "'"); + } + + Map<String, String> properties = Map.of( + "fs.s3a.access.key", creds.accessKeyId(), + "fs.s3a.secret.key", creds.secretAccessKey(), + "fs.s3a.session.token", creds.sessionToken() + ); + + Instant expiration = creds.expiration(); + return new ServiceCredential(properties, expiration); + } catch (SdkException e) { + // SECURITY: Never include the token in exception messages + throw new CredentialResolutionException( + "Failed to assume role '" + cfg.roleArn + "' via AssumeRoleWithWebIdentity: " + + e.getMessage(), e); + } + } + + @Override + public Duration suggestedTtl() { + ResolvedConfig cfg = this.config; + if (cfg != null && cfg.durationSeconds != null) { + return Duration.ofSeconds(cfg.durationSeconds); + } + return Duration.ofMinutes(15); + } + + /** + * Sanitizes a principal string to be valid as an STS role session name. + * STS session names must match [\w+=,.@-]{2,64}. + */ + private static String sanitizeSessionName(String principal) { + String sanitized = principal.replaceAll("[^\\w+=,.@\\-]", "-"); Review Comment: `\\w` in Java regex is locale-independent by default (matches `[a-zA-Z0-9_]`), but using an explicit character class makes the intent clearer and removes any doubt: ```java String sanitized = principal.replaceAll("[^a-zA-Z0-9_+=,.@\\-]", "-"); ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [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]
