[s3-repository] Support IAM roles for Kubernetes service accounts (#81255)

There have been many requests to support repository-s3 authentication via IAM roles in Kubernetes service accounts.

The AWS SDK is supposed to support them out of the box with the aws-java-sdk-sts library. Unfortunately, we can't use WebIdentityTokenCredentialsProvider from the SDK. It reads the token from AWS_WEB_IDENTITY_TOKEN_FILE environment variable which is usually mounted to /var/run/secrets/eks.amazonaws.com/serviceaccount/token and the S3 repository doesn't have the read permission to read it. We don't want to hard-code a file permission for the repository, because the location of AWS_WEB_IDENTITY_TOKEN_FILE can change at any time in the future and we would also generally prefer to restrict the ability of plugins to access things outside of their config directory.

To overcome this limitation, this change adds a custom WebIdentityCredentials provider that reads the service account from a symlink to AWS_WEB_IDENTITY_TOKEN_FILE created in the repository's config directory. We expect the end user to create the symlink to indicate that they want to use service accounts for authentification.

Service accounts are checked and exchanged for session tokens by the AWS STS. To test the authentification flow, this change adds a test fixture which mocks the assume-role-with-web-identity call to the service and returns a response with test credentials.

Fixes #52625

docs/changelog/81255.yaml

@@ -0,0 +1,6 @@
+pr: 81255
+summary: "[s3-repository] Support IAM roles for Kubernetes service accounts"
+area: Snapshot/Restore
+type: feature
+issues:
+ - 52625

docs/reference/snapshot-restore/repository-s3.asciidoc

@@ -12,10 +12,9 @@ https://www.elastic.co/cloud/.*
 To register an S3 repository, specify the type as `s3` when creating
 the repository. The repository defaults to using
 https://docs.aws.amazon.com/AmazonECS/latest/developerguide/task-iam-roles.html[ECS
-IAM Role] or
-https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/iam-roles-for-amazon-ec2.html[EC2
-IAM Role] credentials for authentication. The only mandatory setting is the
-bucket name:
+IAM Role] credentials for authentication. You can also use <<iam-kubernetes-service-accounts>> Kubernetes service accounts.
+
+The only mandatory setting is the bucket name:
 
 [source,console]
 ----
@@ -495,3 +494,24 @@ bandwidth of your VPC's NAT instance.
 
 Instances residing in a public subnet in an AWS VPC will connect to S3 via the
 VPC's internet gateway and not be bandwidth limited by the VPC's NAT instance.
+
+
+[[iam-kubernetes-service-accounts]]
+[discrete]
+==== Using IAM roles for Kubernetes service accounts for authentication
+If you want to use https://aws.amazon.com/blogs/opensource/introducing-fine-grained-iam-roles-service-accounts/[Kubernetes service accounts]
+for authentication, you need to add a symlink to the `$AWS_WEB_IDENTITY_TOKEN_FILE` environment variable
+(which should be automatically set by a Kubernetes pod) in the S3 repository config directory, so the repository
+can have the read access for the service account (a repository can't read any files outside its config directory).
+For example:
+
+[source,bash]
+----
+mkdir -p "${ES_PATH_CONF}/repository-s3"
+ln -s $AWS_WEB_IDENTITY_TOKEN_FILE "${ES_PATH_CONF}/repository-s3/aws-web-identity-token-file"
+----
+
+IMPORTANT: The symlink must be created on all data and master eligible nodes and be readable
+by the `elasticsearch` user. By default, {es} runs as user `elasticsearch` using uid:gid `1000:0`.
+
+If the symlink exists, it will be used by default by all S3 repositories that don't have explicit `client` credentials.

modules/repository-s3/build.gradle

@@ -29,6 +29,7 @@ versions << [
 dependencies {
   api "com.amazonaws:aws-java-sdk-s3:${versions.aws}"
   api "com.amazonaws:aws-java-sdk-core:${versions.aws}"
+  api "com.amazonaws:aws-java-sdk-sts:${versions.aws}"
   api "com.amazonaws:jmespath-java:${versions.aws}"
   api "org.apache.httpcomponents:httpclient:${versions.httpclient}"
   api "org.apache.httpcomponents:httpcore:${versions.httpcore}"
@@ -105,6 +106,9 @@ String s3EC2BasePath = System.getenv("amazon_s3_base_path_ec2")
 String s3ECSBucket = System.getenv("amazon_s3_bucket_ecs")
 String s3ECSBasePath = System.getenv("amazon_s3_base_path_ecs")
 
+String s3STSBucket = System.getenv("amazon_s3_bucket_sts")
+String s3STSBasePath = System.getenv("amazon_s3_base_path_sts")
+
 boolean s3DisableChunkedEncoding = (new Random(Long.parseUnsignedLong(BuildParams.testSeed.tokenize(':').get(0), 16))).nextBoolean()
 
 // If all these variables are missing then we are testing against the internal fixture instead, which has the following
@@ -143,6 +147,13 @@ if (!s3EC2Bucket && !s3EC2BasePath && !s3ECSBucket && !s3ECSBasePath) {
   throw new IllegalArgumentException("not all options specified to run EC2/ECS tests are present")
 }
 
+if (!s3STSBucket && !s3STSBasePath) {
+  s3STSBucket = 'sts_bucket'
+  s3STSBasePath = 'sts_base_path'
+} else if (!s3STSBucket || !s3STSBasePath) {
+  throw new IllegalArgumentException("not all options specified to run STS tests are present")
+}
+
 tasks.named("processYamlRestTestResources").configure {
   Map<String, Object> expansions = [
     'permanent_bucket'        : s3PermanentBucket,
@@ -153,6 +164,8 @@ tasks.named("processYamlRestTestResources").configure {
     'ec2_base_path'           : s3EC2BasePath,
     'ecs_bucket'              : s3ECSBucket,
     'ecs_base_path'           : s3ECSBasePath,
+    'sts_bucket'              : s3STSBucket,
+    'sts_base_path'           : s3STSBasePath,
     'disable_chunked_encoding': s3DisableChunkedEncoding
   ]
   inputs.properties(expansions)
@@ -167,12 +180,14 @@ tasks.named("internalClusterTest").configure {
 tasks.named("yamlRestTest").configure {
     systemProperty 'tests.rest.blacklist', (
       useFixture ?
-        ['repository_s3/50_repository_ecs_credentials/*']
+        ['repository_s3/50_repository_ecs_credentials/*',
+         'repository_s3/60_repository_sts_credentials/*']
         :
         [
           'repository_s3/30_repository_temporary_credentials/*',
           'repository_s3/40_repository_ec2_credentials/*',
-          'repository_s3/50_repository_ecs_credentials/*'
+          'repository_s3/50_repository_ecs_credentials/*',
+          'repository_s3/60_repository_sts_credentials/*'
         ]
     ).join(",")
 }
@@ -226,7 +241,8 @@ if (useFixture) {
     systemProperty 'tests.rest.blacklist', [
       'repository_s3/30_repository_temporary_credentials/*',
       'repository_s3/40_repository_ec2_credentials/*',
-      'repository_s3/50_repository_ecs_credentials/*'
+      'repository_s3/50_repository_ecs_credentials/*',
+      'repository_s3/60_repository_sts_credentials/*'
     ].join(",")
   }
   tasks.named("check").configure { dependsOn("yamlRestTestMinio") }
@@ -253,7 +269,8 @@ if (useFixture) {
       'repository_s3/10_basic/*',
       'repository_s3/20_repository_permanent_credentials/*',
       'repository_s3/30_repository_temporary_credentials/*',
-      'repository_s3/40_repository_ec2_credentials/*'
+      'repository_s3/40_repository_ec2_credentials/*',
+      'repository_s3/60_repository_sts_credentials/*'
     ].join(",")
   }
   tasks.named("check").configure { dependsOn("yamlRestTestECS") }
@@ -265,6 +282,46 @@ if (useFixture) {
   }
 }
 
+// STS (Secure Token Service)
+if (useFixture) {
+  testFixtures.useFixture(':test:fixtures:s3-fixture', 's3-fixture-with-sts')
+  tasks.register("yamlRestTestSTS", RestIntegTestTask.class) {
+    description = "Runs tests with the STS (Secure Token Service)"
+    dependsOn('bundlePlugin')
+
+    SourceSetContainer sourceSets = project.getExtensions().getByType(SourceSetContainer.class);
+    SourceSet yamlRestTestSourceSet = sourceSets.getByName(InternalYamlRestTestPlugin.SOURCE_SET_NAME)
+    setTestClassesDirs(yamlRestTestSourceSet.getOutput().getClassesDirs())
+    setClasspath(yamlRestTestSourceSet.getRuntimeClasspath())
+    systemProperty 'tests.rest.blacklist', [
+      'repository_s3/10_basic/*',
+      'repository_s3/20_repository_permanent_credentials/*',
+      'repository_s3/30_repository_temporary_credentials/*',
+      'repository_s3/40_repository_ec2_credentials/*',
+      'repository_s3/50_repository_ecs_credentials/*'
+    ].join(",")
+  }
+  tasks.named("check").configure { dependsOn("yamlRestTestSTS") }
+
+  testClusters.matching { it.name == "yamlRestTestSTS" }.configureEach {
+    module tasks.bundlePlugin.archiveFile
+
+    setting 's3.client.integration_test_sts.endpoint', { "${-> fixtureAddress('s3-fixture', 's3-fixture-with-sts', '80')}" }, IGNORE_VALUE
+    systemProperty 'com.amazonaws.sdk.stsMetadataServiceEndpointOverride',
+      { "${-> fixtureAddress('s3-fixture', 's3-fixture-with-sts', '80')}/assume-role-with-web-identity" }, IGNORE_VALUE
+
+    File awsWebIdentityTokenExternalLocation = file('src/test/resources/aws-web-identity-token-file')
+    // The web identity token can be read only from the plugin config directory because of security restrictions
+    // Ideally we would create a symlink, but extraConfigFile doesn't support it
+    extraConfigFile 'repository-s3/aws-web-identity-token-file', awsWebIdentityTokenExternalLocation
+    environment 'AWS_WEB_IDENTITY_TOKEN_FILE', "$awsWebIdentityTokenExternalLocation"
+
+    // The AWS STS SDK requires the role and session names to be set. We can verify that they are sent to S3S in the S3HttpFixtureWithSTS fixture
+    environment 'AWS_ROLE_ARN', 'arn:aws:iam::123456789012:role/FederatedWebIdentityRole'
+    environment 'AWS_ROLE_SESSION_NAME', 'sts-fixture-test'
+  }
+}
+
 // 3rd Party Tests
 TaskProvider s3ThirdPartyTest = tasks.register("s3ThirdPartyTest", Test) {
   SourceSetContainer sourceSets = project.getExtensions().getByType(SourceSetContainer.class);

modules/repository-s3/licenses/aws-java-sdk-sts-1.11.749.jar.sha1

@@ -0,0 +1 @@
+724bd22c0ff41c496469e18f9bea12bdfb2f7540

modules/repository-s3/src/internalClusterTest/java/org/elasticsearch/repositories/s3/S3BlobStoreRepositoryTests.java

@@ -226,7 +226,7 @@ public class S3BlobStoreRepositoryTests extends ESMockAPIBasedRepositoryIntegTes
             BigArrays bigArrays,
             RecoverySettings recoverySettings
         ) {
-            return new S3Repository(metadata, registry, service, clusterService, bigArrays, recoverySettings) {
+            return new S3Repository(metadata, registry, getService(), clusterService, bigArrays, recoverySettings) {
 
                 @Override
                 public BlobStore blobStore() {

modules/repository-s3/src/main/java/org/elasticsearch/repositories/s3/S3RepositoryPlugin.java

@@ -10,28 +10,38 @@ package org.elasticsearch.repositories.s3;
 
 import com.amazonaws.util.json.Jackson;
 
+import org.apache.lucene.util.SetOnce;
 import org.elasticsearch.SpecialPermission;
+import org.elasticsearch.client.internal.Client;
+import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.metadata.RepositoryMetadata;
 import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.BigArrays;
 import org.elasticsearch.env.Environment;
+import org.elasticsearch.env.NodeEnvironment;
 import org.elasticsearch.indices.recovery.RecoverySettings;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.plugins.ReloadablePlugin;
 import org.elasticsearch.plugins.RepositoryPlugin;
+import org.elasticsearch.repositories.RepositoriesService;
 import org.elasticsearch.repositories.Repository;
+import org.elasticsearch.script.ScriptService;
+import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.watcher.ResourceWatcherService;
 import org.elasticsearch.xcontent.NamedXContentRegistry;
 
 import java.io.IOException;
 import java.security.AccessController;
 import java.security.PrivilegedAction;
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
-import java.util.Objects;
+import java.util.function.Supplier;
 
 /**
  * A plugin to add a repository type that writes to and from the AWS S3.
@@ -54,17 +64,15 @@ public class S3RepositoryPlugin extends Plugin implements RepositoryPlugin, Relo
         });
     }
 
-    protected final S3Service service;
+    private final SetOnce<S3Service> service = new SetOnce<>();
+    private final Settings settings;
 
-    public S3RepositoryPlugin(final Settings settings) {
-        this(settings, new S3Service());
+    public S3RepositoryPlugin(Settings settings) {
+        this.settings = settings;
     }
 
-    S3RepositoryPlugin(final Settings settings, final S3Service service) {
-        this.service = Objects.requireNonNull(service, "S3 service must not be null");
-        // eagerly load client settings so that secure settings are read
-        final Map<String, S3ClientSettings> clientsSettings = S3ClientSettings.load(settings);
-        this.service.refreshAndClearCache(clientsSettings);
+    S3Service getService() {
+        return service.get();
     }
 
     // proxy method for testing
@@ -75,7 +83,30 @@ public class S3RepositoryPlugin extends Plugin implements RepositoryPlugin, Relo
         final BigArrays bigArrays,
         final RecoverySettings recoverySettings
     ) {
-        return new S3Repository(metadata, registry, service, clusterService, bigArrays, recoverySettings);
+        return new S3Repository(metadata, registry, service.get(), clusterService, bigArrays, recoverySettings);
+    }
+
+    @Override
+    public Collection<Object> createComponents(
+        Client client,
+        ClusterService clusterService,
+        ThreadPool threadPool,
+        ResourceWatcherService resourceWatcherService,
+        ScriptService scriptService,
+        NamedXContentRegistry xContentRegistry,
+        Environment environment,
+        NodeEnvironment nodeEnvironment,
+        NamedWriteableRegistry namedWriteableRegistry,
+        IndexNameExpressionResolver indexNameExpressionResolver,
+        Supplier<RepositoriesService> repositoriesServiceSupplier
+    ) {
+        service.set(s3Service(environment));
+        this.service.get().refreshAndClearCache(S3ClientSettings.load(settings));
+        return List.of(service);
+    }
+
+    S3Service s3Service(Environment environment) {
+        return new S3Service(environment);
     }
 
     @Override
@@ -118,11 +149,11 @@ public class S3RepositoryPlugin extends Plugin implements RepositoryPlugin, Relo
     public void reload(Settings settings) {
         // secure settings should be readable
         final Map<String, S3ClientSettings> clientsSettings = S3ClientSettings.load(settings);
-        service.refreshAndClearCache(clientsSettings);
+        getService().refreshAndClearCache(clientsSettings);
     }
 
     @Override
     public void close() throws IOException {
-        service.close();
+        getService().close();
     }
 }

modules/repository-s3/src/main/java/org/elasticsearch/repositories/s3/S3Service.java

@@ -11,13 +11,19 @@ package org.elasticsearch.repositories.s3;
 import com.amazonaws.ClientConfiguration;
 import com.amazonaws.auth.AWSCredentials;
 import com.amazonaws.auth.AWSCredentialsProvider;
+import com.amazonaws.auth.AWSCredentialsProviderChain;
 import com.amazonaws.auth.AWSStaticCredentialsProvider;
+import com.amazonaws.auth.AnonymousAWSCredentials;
 import com.amazonaws.auth.EC2ContainerCredentialsProviderWrapper;
+import com.amazonaws.auth.STSAssumeRoleWithWebIdentitySessionCredentialsProvider;
 import com.amazonaws.client.builder.AwsClientBuilder;
 import com.amazonaws.http.IdleConnectionReaper;
 import com.amazonaws.services.s3.AmazonS3;
 import com.amazonaws.services.s3.AmazonS3ClientBuilder;
 import com.amazonaws.services.s3.internal.Constants;
+import com.amazonaws.services.securitytoken.AWSSecurityTokenService;
+import com.amazonaws.services.securitytoken.AWSSecurityTokenServiceClient;
+import com.amazonaws.services.securitytoken.AWSSecurityTokenServiceClientBuilder;
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
@@ -25,10 +31,19 @@ import org.elasticsearch.cluster.metadata.RepositoryMetadata;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.Maps;
+import org.elasticsearch.core.internal.io.IOUtils;
+import org.elasticsearch.env.Environment;
 
 import java.io.Closeable;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
 import java.util.Map;
+import java.util.Objects;
 
+import static com.amazonaws.SDKGlobalConfiguration.AWS_ROLE_ARN_ENV_VAR;
+import static com.amazonaws.SDKGlobalConfiguration.AWS_ROLE_SESSION_NAME_ENV_VAR;
+import static com.amazonaws.SDKGlobalConfiguration.AWS_WEB_IDENTITY_ENV_VAR;
 import static java.util.Collections.emptyMap;
 
 class S3Service implements Closeable {
@@ -50,6 +65,12 @@ class S3Service implements Closeable {
      */
     private volatile Map<Settings, S3ClientSettings> derivedClientSettings = emptyMap();
 
+    final CustomWebIdentityTokenCredentialsProvider webIdentityTokenCredentialsProvider;
+
+    S3Service(Environment environment) {
+        webIdentityTokenCredentialsProvider = new CustomWebIdentityTokenCredentialsProvider(environment);
+    }
+
     /**
      * Refreshes the settings for the AmazonS3 clients and clears the cache of
      * existing clients. New clients will be build using these new settings. Old
@@ -128,7 +149,7 @@ class S3Service implements Closeable {
     // proxy for testing
     AmazonS3 buildClient(final S3ClientSettings clientSettings) {
         final AmazonS3ClientBuilder builder = AmazonS3ClientBuilder.standard();
-        builder.withCredentials(buildCredentials(LOGGER, clientSettings));
+        builder.withCredentials(buildCredentials(LOGGER, clientSettings, webIdentityTokenCredentialsProvider));
         builder.withClientConfiguration(buildConfiguration(clientSettings));
 
         String endpoint = Strings.hasLength(clientSettings.endpoint) ? clientSettings.endpoint : Constants.S3_HOSTNAME;
@@ -186,11 +207,22 @@ class S3Service implements Closeable {
     }
 
     // pkg private for tests
-    static AWSCredentialsProvider buildCredentials(Logger logger, S3ClientSettings clientSettings) {
+    static AWSCredentialsProvider buildCredentials(
+        Logger logger,
+        S3ClientSettings clientSettings,
+        CustomWebIdentityTokenCredentialsProvider webIdentityTokenCredentialsProvider
+    ) {
         final S3BasicCredentials credentials = clientSettings.credentials;
         if (credentials == null) {
-            logger.debug("Using instance profile credentials");
-            return new PrivilegedInstanceProfileCredentialsProvider();
+            if (webIdentityTokenCredentialsProvider.isActive()) {
+                logger.debug("Using a custom provider chain of Web Identity Token and instance profile credentials");
+                return new PrivilegedAWSCredentialsProvider(
+                    new AWSCredentialsProviderChain(webIdentityTokenCredentialsProvider, new EC2ContainerCredentialsProviderWrapper())
+                );
+            } else {
+                logger.debug("Using instance profile credentials");
+                return new PrivilegedAWSCredentialsProvider(new EC2ContainerCredentialsProviderWrapper());
+            }
         } else {
             logger.debug("Using basic key/secret credentials");
             return new AWSStaticCredentialsProvider(credentials);
@@ -210,27 +242,114 @@ class S3Service implements Closeable {
         IdleConnectionReaper.shutdown();
     }
 
-    static class PrivilegedInstanceProfileCredentialsProvider implements AWSCredentialsProvider {
-        private final AWSCredentialsProvider credentials;
+    @Override
+    public void close() throws IOException {
+        releaseCachedClients();
+        webIdentityTokenCredentialsProvider.shutdown();
+    }
+
+    static class PrivilegedAWSCredentialsProvider implements AWSCredentialsProvider {
+        private final AWSCredentialsProvider credentialsProvider;
+
+        private PrivilegedAWSCredentialsProvider(AWSCredentialsProvider credentialsProvider) {
+            this.credentialsProvider = credentialsProvider;
+        }
 
-        private PrivilegedInstanceProfileCredentialsProvider() {
-            // InstanceProfileCredentialsProvider as last item of chain
-            this.credentials = new EC2ContainerCredentialsProviderWrapper();
+        AWSCredentialsProvider getCredentialsProvider() {
+            return credentialsProvider;
         }
 
         @Override
         public AWSCredentials getCredentials() {
-            return SocketAccess.doPrivileged(credentials::getCredentials);
+            return SocketAccess.doPrivileged(credentialsProvider::getCredentials);
         }
 
         @Override
         public void refresh() {
-            SocketAccess.doPrivilegedVoid(credentials::refresh);
+            SocketAccess.doPrivilegedVoid(credentialsProvider::refresh);
         }
     }
 
-    @Override
-    public void close() {
-        releaseCachedClients();
+    /**
+     * Customizes {@link com.amazonaws.auth.WebIdentityTokenCredentialsProvider}
+     *
+     * <ul>
+     * <li>Reads the the location of the web identity token not from AWS_WEB_IDENTITY_TOKEN_FILE, but from a symlink
+     * in the plugin directory, so we don't need to create a hardcoded read file permission for the plugin.</li>
+     * <li>Supports customization of the STS endpoint via a system property, so we can test it against a test fixture.</li>
+     * <li>Supports gracefully shutting down the provider and the STS client.</li>
+     * </ul>
+     */
+    static class CustomWebIdentityTokenCredentialsProvider implements AWSCredentialsProvider {
+
+        private STSAssumeRoleWithWebIdentitySessionCredentialsProvider credentialsProvider;
+        private AWSSecurityTokenService stsClient;
+
+        CustomWebIdentityTokenCredentialsProvider(Environment environment) {
+            // Check whether the original environment variable exists. If it doesn't,
+            // the system doesn't support AWS web identity tokens
+            if (System.getenv(AWS_WEB_IDENTITY_ENV_VAR) == null) {
+                return;
+            }
+            // Make sure that a readable symlink to the token file exists in the plugin config directory
+            Path webIdentityTokenFileSymlink = environment.configFile().resolve("repository-s3/aws-web-identity-token-file");
+            if (Files.exists(webIdentityTokenFileSymlink) == false) {
+                throw new IllegalStateException("A Web Identity Token symlink in the config directory doesn't exist");
+            }
+            if (Files.isReadable(webIdentityTokenFileSymlink) == false) {
+                throw new IllegalStateException("Unable to read a Web Identity Token symlink in the config directory");
+            }
+            String roleArn = System.getenv(AWS_ROLE_ARN_ENV_VAR);
+            String roleSessionName = System.getenv(AWS_ROLE_SESSION_NAME_ENV_VAR);
+            if (roleArn == null || roleSessionName == null) {
+                LOGGER.warn(
+                    "Unable to use a web identity token for authentication. The AWS_WEB_IDENTITY_TOKEN_FILE environment "
+                        + "variable is set, but either AWS_ROLE_ARN or AWS_ROLE_SESSION_NAME are missing"
+                );
+                return;
+            }
+            AWSSecurityTokenServiceClientBuilder stsClientBuilder = AWSSecurityTokenServiceClient.builder();
+
+            // Just for testing
+            String customStsEndpoint = System.getProperty("com.amazonaws.sdk.stsMetadataServiceEndpointOverride");
+            if (customStsEndpoint != null) {
+                stsClientBuilder.withEndpointConfiguration(new AwsClientBuilder.EndpointConfiguration(customStsEndpoint, null));
+            }
+            stsClientBuilder.withCredentials(new AWSStaticCredentialsProvider(new AnonymousAWSCredentials()));
+            stsClient = SocketAccess.doPrivileged(stsClientBuilder::build);
+            try {
+                credentialsProvider = new STSAssumeRoleWithWebIdentitySessionCredentialsProvider.Builder(
+                    roleArn,
+                    roleSessionName,
+                    webIdentityTokenFileSymlink.toString()
+                ).withStsClient(stsClient).build();
+            } catch (Exception e) {
+                stsClient.shutdown();
+                throw e;
+            }
+        }
+
+        boolean isActive() {
+            return credentialsProvider != null;
+        }
+
+        @Override
+        public AWSCredentials getCredentials() {
+            Objects.requireNonNull(credentialsProvider, "credentialsProvider is not set");
+            return credentialsProvider.getCredentials();
+        }
+
+        @Override
+        public void refresh() {
+            if (credentialsProvider != null) {
+                credentialsProvider.refresh();
+            }
+        }
+
+        public void shutdown() throws IOException {
+            if (credentialsProvider != null) {
+                IOUtils.close(credentialsProvider, () -> stsClient.shutdown());
+            }
+        }
     }
 }

modules/repository-s3/src/test/java/org/elasticsearch/repositories/s3/AwsS3ServiceImplTests.java

@@ -10,12 +10,17 @@ package org.elasticsearch.repositories.s3;
 
 import com.amazonaws.ClientConfiguration;
 import com.amazonaws.Protocol;
+import com.amazonaws.auth.AWSCredentials;
 import com.amazonaws.auth.AWSCredentialsProvider;
+import com.amazonaws.auth.AWSCredentialsProviderChain;
 import com.amazonaws.auth.AWSStaticCredentialsProvider;
+import com.amazonaws.auth.BasicAWSCredentials;
+import com.amazonaws.auth.EC2ContainerCredentialsProviderWrapper;
 
 import org.elasticsearch.common.settings.MockSecureSettings;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.test.ESTestCase;
+import org.mockito.Mockito;
 
 import java.util.Locale;
 import java.util.Map;
@@ -26,11 +31,39 @@ import static org.hamcrest.Matchers.is;
 
 public class AwsS3ServiceImplTests extends ESTestCase {
 
+    private final S3Service.CustomWebIdentityTokenCredentialsProvider webIdentityTokenCredentialsProvider = Mockito.mock(
+        S3Service.CustomWebIdentityTokenCredentialsProvider.class
+    );
+
     public void testAWSCredentialsDefaultToInstanceProviders() {
         final String inexistentClientName = randomAlphaOfLength(8).toLowerCase(Locale.ROOT);
         final S3ClientSettings clientSettings = S3ClientSettings.getClientSettings(Settings.EMPTY, inexistentClientName);
-        final AWSCredentialsProvider credentialsProvider = S3Service.buildCredentials(logger, clientSettings);
-        assertThat(credentialsProvider, instanceOf(S3Service.PrivilegedInstanceProfileCredentialsProvider.class));
+        final AWSCredentialsProvider credentialsProvider = S3Service.buildCredentials(
+            logger,
+            clientSettings,
+            webIdentityTokenCredentialsProvider
+        );
+        assertThat(credentialsProvider, instanceOf(S3Service.PrivilegedAWSCredentialsProvider.class));
+        var privilegedAWSCredentialsProvider = (S3Service.PrivilegedAWSCredentialsProvider) credentialsProvider;
+        assertThat(privilegedAWSCredentialsProvider.getCredentialsProvider(), instanceOf(EC2ContainerCredentialsProviderWrapper.class));
+    }
+
+    public void testSupportsWebIdentityTokenCredentials() {
+        Mockito.when(webIdentityTokenCredentialsProvider.getCredentials())
+            .thenReturn(new BasicAWSCredentials("sts_access_key_id", "sts_secret_key"));
+        Mockito.when(webIdentityTokenCredentialsProvider.isActive()).thenReturn(true);
+
+        AWSCredentialsProvider credentialsProvider = S3Service.buildCredentials(
+            logger,
+            S3ClientSettings.getClientSettings(Settings.EMPTY, randomAlphaOfLength(8).toLowerCase(Locale.ROOT)),
+            webIdentityTokenCredentialsProvider
+        );
+        assertThat(credentialsProvider, instanceOf(S3Service.PrivilegedAWSCredentialsProvider.class));
+        var privilegedAWSCredentialsProvider = (S3Service.PrivilegedAWSCredentialsProvider) credentialsProvider;
+        assertThat(privilegedAWSCredentialsProvider.getCredentialsProvider(), instanceOf(AWSCredentialsProviderChain.class));
+        AWSCredentials credentials = privilegedAWSCredentialsProvider.getCredentials();
+        assertEquals("sts_access_key_id", credentials.getAWSAccessKeyId());
+        assertEquals("sts_secret_key", credentials.getAWSSecretKey());
     }
 
     public void testAWSCredentialsFromKeystore() {
@@ -49,15 +82,25 @@ public class AwsS3ServiceImplTests extends ESTestCase {
         for (int i = 0; i < clientsCount; i++) {
             final String clientName = clientNamePrefix + i;
             final S3ClientSettings someClientSettings = allClientsSettings.get(clientName);
-            final AWSCredentialsProvider credentialsProvider = S3Service.buildCredentials(logger, someClientSettings);
+            final AWSCredentialsProvider credentialsProvider = S3Service.buildCredentials(
+                logger,
+                someClientSettings,
+                webIdentityTokenCredentialsProvider
+            );
             assertThat(credentialsProvider, instanceOf(AWSStaticCredentialsProvider.class));
             assertThat(credentialsProvider.getCredentials().getAWSAccessKeyId(), is(clientName + "_aws_access_key"));
             assertThat(credentialsProvider.getCredentials().getAWSSecretKey(), is(clientName + "_aws_secret_key"));
         }
         // test default exists and is an Instance provider
         final S3ClientSettings defaultClientSettings = allClientsSettings.get("default");
-        final AWSCredentialsProvider defaultCredentialsProvider = S3Service.buildCredentials(logger, defaultClientSettings);
-        assertThat(defaultCredentialsProvider, instanceOf(S3Service.PrivilegedInstanceProfileCredentialsProvider.class));
+        final AWSCredentialsProvider defaultCredentialsProvider = S3Service.buildCredentials(
+            logger,
+            defaultClientSettings,
+            webIdentityTokenCredentialsProvider
+        );
+        assertThat(defaultCredentialsProvider, instanceOf(S3Service.PrivilegedAWSCredentialsProvider.class));
+        var privilegedAWSCredentialsProvider = (S3Service.PrivilegedAWSCredentialsProvider) defaultCredentialsProvider;
+        assertThat(privilegedAWSCredentialsProvider.getCredentialsProvider(), instanceOf(EC2ContainerCredentialsProviderWrapper.class));
     }
 
     public void testSetDefaultCredential() {
@@ -71,7 +114,11 @@ public class AwsS3ServiceImplTests extends ESTestCase {
         assertThat(allClientsSettings.size(), is(1));
         // test default exists and is an Instance provider
         final S3ClientSettings defaultClientSettings = allClientsSettings.get("default");
-        final AWSCredentialsProvider defaultCredentialsProvider = S3Service.buildCredentials(logger, defaultClientSettings);
+        final AWSCredentialsProvider defaultCredentialsProvider = S3Service.buildCredentials(
+            logger,
+            defaultClientSettings,
+            webIdentityTokenCredentialsProvider
+        );
         assertThat(defaultCredentialsProvider, instanceOf(AWSStaticCredentialsProvider.class));
         assertThat(defaultCredentialsProvider.getCredentials().getAWSAccessKeyId(), is(awsAccessKey));
         assertThat(defaultCredentialsProvider.getCredentials().getAWSSecretKey(), is(awsSecretKey));

modules/repository-s3/src/test/java/org/elasticsearch/repositories/s3/RepositoryCredentialsTests.java

@@ -19,6 +19,7 @@ import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.settings.MockSecureSettings;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.BigArrays;
+import org.elasticsearch.env.Environment;
 import org.elasticsearch.indices.recovery.RecoverySettings;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.plugins.PluginsService;
@@ -134,7 +135,7 @@ public class RepositoryCredentialsTests extends ESSingleNodeTestCase {
     public static final class ProxyS3RepositoryPlugin extends S3RepositoryPlugin {
 
         public ProxyS3RepositoryPlugin(Settings settings) {
-            super(settings, new ProxyS3Service());
+            super(settings);
         }
 
         @Override
@@ -145,7 +146,7 @@ public class RepositoryCredentialsTests extends ESSingleNodeTestCase {
             BigArrays bigArrays,
             RecoverySettings recoverySettings
         ) {
-            return new S3Repository(metadata, registry, service, clusterService, bigArrays, recoverySettings) {
+            return new S3Repository(metadata, registry, getService(), clusterService, bigArrays, recoverySettings) {
                 @Override
                 protected void assertSnapshotOrGenericThread() {
                     // eliminate thread name check as we create repo manually on test/main threads
@@ -153,6 +154,11 @@ public class RepositoryCredentialsTests extends ESSingleNodeTestCase {
             };
         }
 
+        @Override
+        S3Service s3Service(Environment environment) {
+            return new ProxyS3Service(environment);
+        }
+
         public static final class ClientAndCredentials extends AmazonS3Wrapper {
             final AWSCredentialsProvider credentials;
 
@@ -166,10 +172,14 @@ public class RepositoryCredentialsTests extends ESSingleNodeTestCase {
 
             private static final Logger logger = LogManager.getLogger(ProxyS3Service.class);
 
+            ProxyS3Service(Environment environment) {
+                super(environment);
+            }
+
             @Override
             AmazonS3 buildClient(final S3ClientSettings clientSettings) {
                 final AmazonS3 client = super.buildClient(clientSettings);
-                return new ClientAndCredentials(client, buildCredentials(logger, clientSettings));
+                return new ClientAndCredentials(client, buildCredentials(logger, clientSettings, webIdentityTokenCredentialsProvider));
             }
 
         }

modules/repository-s3/src/test/java/org/elasticsearch/repositories/s3/S3BlobContainerRetriesTests.java

@@ -30,9 +30,11 @@ import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.SuppressForbidden;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.core.internal.io.IOUtils;
+import org.elasticsearch.env.Environment;
 import org.elasticsearch.repositories.blobstore.AbstractBlobContainerRetriesTestCase;
 import org.junit.After;
 import org.junit.Before;
+import org.mockito.Mockito;
 
 import java.io.ByteArrayInputStream;
 import java.io.FilterInputStream;
@@ -66,7 +68,7 @@ public class S3BlobContainerRetriesTests extends AbstractBlobContainerRetriesTes
 
     @Before
     public void setUp() throws Exception {
-        service = new S3Service();
+        service = new S3Service(Mockito.mock(Environment.class));
         super.setUp();
     }
 

modules/repository-s3/src/test/java/org/elasticsearch/repositories/s3/S3ClientSettingsTests.java

@@ -14,8 +14,11 @@ import com.amazonaws.services.s3.AmazonS3Client;
 
 import org.elasticsearch.common.settings.MockSecureSettings;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.env.Environment;
 import org.elasticsearch.test.ESTestCase;
+import org.mockito.Mockito;
 
+import java.io.IOException;
 import java.util.Map;
 
 import static org.hamcrest.Matchers.contains;
@@ -167,14 +170,14 @@ public class S3ClientSettingsTests extends ESTestCase {
         assertThat(settings.get("other").disableChunkedEncoding, is(true));
     }
 
-    public void testRegionCanBeSet() {
+    public void testRegionCanBeSet() throws IOException {
         final String region = randomAlphaOfLength(5);
         final Map<String, S3ClientSettings> settings = S3ClientSettings.load(
             Settings.builder().put("s3.client.other.region", region).build()
         );
         assertThat(settings.get("default").region, is(""));
         assertThat(settings.get("other").region, is(region));
-        try (S3Service s3Service = new S3Service()) {
+        try (S3Service s3Service = new S3Service(Mockito.mock(Environment.class))) {
             AmazonS3Client other = (AmazonS3Client) s3Service.buildClient(settings.get("other"));
             assertThat(other.getSignerRegionOverride(), is(region));
         }

modules/repository-s3/src/test/java/org/elasticsearch/repositories/s3/S3RepositoryTests.java

@@ -16,12 +16,14 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.ByteSizeUnit;
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.util.MockBigArrays;
+import org.elasticsearch.env.Environment;
 import org.elasticsearch.indices.recovery.RecoverySettings;
 import org.elasticsearch.repositories.RepositoryException;
 import org.elasticsearch.repositories.blobstore.BlobStoreTestUtil;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.xcontent.NamedXContentRegistry;
 import org.hamcrest.Matchers;
+import org.mockito.Mockito;
 
 import java.util.Map;
 
@@ -41,6 +43,11 @@ public class S3RepositoryTests extends ESTestCase {
     }
 
     private static class DummyS3Service extends S3Service {
+
+        DummyS3Service(Environment environment) {
+            super(environment);
+        }
+
         @Override
         public AmazonS3Reference client(RepositoryMetadata repositoryMetadata) {
             return new AmazonS3Reference(new DummyS3Client());
@@ -117,7 +124,7 @@ public class S3RepositoryTests extends ESTestCase {
         return new S3Repository(
             metadata,
             NamedXContentRegistry.EMPTY,
-            new DummyS3Service(),
+            new DummyS3Service(Mockito.mock(Environment.class)),
             BlobStoreTestUtil.mockClusterService(),
             MockBigArrays.NON_RECYCLING_INSTANCE,
             new RecoverySettings(Settings.EMPTY, new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS))

modules/repository-s3/src/test/java/org/elasticsearch/repositories/s3/S3ServiceTests.java

@@ -9,12 +9,16 @@ package org.elasticsearch.repositories.s3;
 
 import org.elasticsearch.cluster.metadata.RepositoryMetadata;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.env.Environment;
 import org.elasticsearch.test.ESTestCase;
+import org.mockito.Mockito;
+
+import java.io.IOException;
 
 public class S3ServiceTests extends ESTestCase {
 
-    public void testCachedClientsAreReleased() {
-        final S3Service s3Service = new S3Service();
+    public void testCachedClientsAreReleased() throws IOException {
+        final S3Service s3Service = new S3Service(Mockito.mock(Environment.class));
         final Settings settings = Settings.builder().put("endpoint", "http://first").build();
         final RepositoryMetadata metadata1 = new RepositoryMetadata("first", "s3", settings);
         final RepositoryMetadata metadata2 = new RepositoryMetadata("second", "s3", settings);

modules/repository-s3/src/test/resources/aws-web-identity-token-file

@@ -0,0 +1 @@
+Atza|IQEBLjAsAhRFiXuWpUXuRvQ9PZL3GMFcYevydwIUFAHZwXZXXXXXXXXJnrulxKDHwy87oGKPznh0D6bEQZTSCzyoCtL_8S07pLpr0zMbn6w1lfVZKNTBdDansFBmtGnIsIapjI6xKR02Yc_2bQ8LZbUXSGm6Ry6_BG7PrtLZtj_dfCTj92xNGed-CrKqjG7nPBjNIL016GGvuS5gSvPRUxWES3VYfm1wl7WTI7jn-Pcb6M-buCgHhFOzTQxod27L9CqnOLio7N3gZAGpsp6n1-AJBOCJckcyXe2c6uD0srOJeZlKUm2eTDVMf8IehDVI0r1QOnTV6KzzAI3OY87Vd_cVMQ

modules/repository-s3/src/yamlRestTest/resources/rest-api-spec/test/repository_s3/60_repository_sts_credentials.yml

@@ -0,0 +1,266 @@
+# Integration tests for repository-s3
+
+---
+setup:
+
+  # Register repository with sts credentials
+  - do:
+      snapshot.create_repository:
+        repository: repository_sts
+        body:
+          type: s3
+          settings:
+            bucket: @sts_bucket@
+            client: integration_test_sts
+            base_path: "@sts_base_path@"
+            canned_acl: private
+            storage_class: standard
+            disable_chunked_encoding: @disable_chunked_encoding@
+
+---
+"Snapshot and Restore repository-s3 using sts credentials":
+
+  # Get repository
+  - do:
+      snapshot.get_repository:
+        repository: repository_sts
+
+  - match: { repository_sts.settings.bucket: @sts_bucket@ }
+  - match: { repository_sts.settings.client: "integration_test_sts" }
+  - match: { repository_sts.settings.base_path: "@sts_base_path@" }
+  - match: { repository_sts.settings.canned_acl: "private" }
+  - match: { repository_sts.settings.storage_class: "standard" }
+  - is_false: repository_sts.settings.access_key
+  - is_false: repository_sts.settings.secret_key
+  - is_false: repository_sts.settings.session_token
+
+  # Index documents
+  - do:
+      bulk:
+        refresh: true
+        body:
+          - index:
+              _index: docs
+              _id: 1
+          - snapshot: one
+          - index:
+              _index: docs
+              _id: 2
+          - snapshot: one
+          - index:
+              _index: docs
+              _id: 3
+          - snapshot: one
+
+  - do:
+      count:
+        index: docs
+
+  - match: { count: 3 }
+
+  # Create a first snapshot
+  - do:
+      snapshot.create:
+        repository: repository_sts
+        snapshot: snapshot-one
+        wait_for_completion: true
+
+  - match: { snapshot.snapshot: snapshot-one }
+  - match: { snapshot.state: SUCCESS }
+  - match: { snapshot.include_global_state: true }
+  - match: { snapshot.shards.failed: 0 }
+
+  - do:
+      snapshot.status:
+        repository: repository_sts
+        snapshot: snapshot-one
+
+  - is_true: snapshots
+  - match: { snapshots.0.snapshot: snapshot-one }
+  - match: { snapshots.0.state: SUCCESS }
+
+  # Index more documents
+  - do:
+      bulk:
+        refresh: true
+        body:
+          - index:
+              _index: docs
+              _id: 4
+          - snapshot: two
+          - index:
+              _index: docs
+              _id: 5
+          - snapshot: two
+          - index:
+              _index: docs
+              _id: 6
+          - snapshot: two
+          - index:
+              _index: docs
+              _id: 7
+          - snapshot: two
+
+  - do:
+      count:
+        index: docs
+
+  - match: { count: 7 }
+
+  # Create a second snapshot
+  - do:
+      snapshot.create:
+        repository: repository_sts
+        snapshot: snapshot-two
+        wait_for_completion: true
+
+  - match: { snapshot.snapshot: snapshot-two }
+  - match: { snapshot.state: SUCCESS }
+  - match: { snapshot.shards.failed: 0 }
+
+  - do:
+      snapshot.get:
+        repository: repository_sts
+        snapshot: snapshot-one,snapshot-two
+
+  - is_true: snapshots
+  - match: { snapshots.0.state: SUCCESS }
+  - match: { snapshots.1.state: SUCCESS }
+
+  # Delete the index
+  - do:
+      indices.delete:
+        index: docs
+
+  # Restore the second snapshot
+  - do:
+      snapshot.restore:
+        repository: repository_sts
+        snapshot: snapshot-two
+        wait_for_completion: true
+
+  - do:
+      count:
+        index: docs
+
+  - match: { count: 7 }
+
+  # Delete the index again
+  - do:
+      indices.delete:
+        index: docs
+
+  # Restore the first snapshot
+  - do:
+      snapshot.restore:
+        repository: repository_sts
+        snapshot: snapshot-one
+        wait_for_completion: true
+
+  - do:
+      count:
+        index: docs
+
+  - match: { count: 3 }
+
+  # Remove the snapshots
+  - do:
+      snapshot.delete:
+        repository: repository_sts
+        snapshot: snapshot-two
+
+  - do:
+      snapshot.delete:
+        repository: repository_sts
+        snapshot: snapshot-one
+
+---
+
+"Register a repository with a non existing bucket":
+
+  - do:
+      catch: /repository_verification_exception/
+      snapshot.create_repository:
+        repository: repository_sts
+        body:
+          type: s3
+          settings:
+            bucket: zHHkfSqlbnBsbpSgvCYtxrEfFLqghXtyPvvvKPNBnRCicNHQLE
+            client: integration_test_sts
+
+---
+"Register a repository with a non existing client":
+
+  - do:
+      catch: /illegal_argument_exception/
+      snapshot.create_repository:
+        repository: repository_sts
+        body:
+          type: s3
+          settings:
+            bucket: repository_sts
+            client: unknown
+
+---
+"Register a read-only repository with a non existing bucket":
+
+  - do:
+      catch: /repository_verification_exception/
+      snapshot.create_repository:
+        repository: repository_sts
+        body:
+          type: s3
+          settings:
+            readonly: true
+            bucket: zHHkfSqlbnBsbpSgvCYtxrEfFLqghXtyPvvvKPNBnRCicNHQLE
+            client: integration_test_sts
+
+---
+"Register a read-only repository with a non existing client":
+
+  - do:
+      catch: /illegal_argument_exception/
+      snapshot.create_repository:
+        repository: repository_sts
+        body:
+          type: s3
+          settings:
+            readonly: true
+            bucket: repository_sts
+            client: unknown
+
+---
+"Get a non existing snapshot":
+
+  - do:
+      catch: /snapshot_missing_exception/
+      snapshot.get:
+        repository: repository_sts
+        snapshot: missing
+
+---
+"Delete a non existing snapshot":
+
+  - do:
+      catch: /snapshot_missing_exception/
+      snapshot.delete:
+        repository: repository_sts
+        snapshot: missing
+
+---
+"Restore a non existing snapshot":
+
+  - do:
+      catch: /snapshot_restore_exception/
+      snapshot.restore:
+        repository: repository_sts
+        snapshot: missing
+        wait_for_completion: true
+
+---
+teardown:
+
+  # Remove our repository
+  - do:
+      snapshot.delete_repository:
+        repository: repository_sts

test/fixtures/s3-fixture/docker-compose.yml

@@ -122,3 +122,20 @@ services:
           - ./testfixtures_shared/shared:/fixture/shared
       ports:
           - "80"
+
+  s3-fixture-with-sts:
+    build:
+      context: .
+      args:
+        fixtureClass: fixture.s3.S3HttpFixtureWithSTS
+        port: 80
+        bucket: "sts_bucket"
+        basePath: "sts_base_path"
+        accessKey: "sts_access_key"
+        sessionToken: "sts_session_token"
+        webIdentityToken: "Atza|IQEBLjAsAhRFiXuWpUXuRvQ9PZL3GMFcYevydwIUFAHZwXZXXXXXXXXJnrulxKDHwy87oGKPznh0D6bEQZTSCzyoCtL_8S07pLpr0zMbn6w1lfVZKNTBdDansFBmtGnIsIapjI6xKR02Yc_2bQ8LZbUXSGm6Ry6_BG7PrtLZtj_dfCTj92xNGed-CrKqjG7nPBjNIL016GGvuS5gSvPRUxWES3VYfm1wl7WTI7jn-Pcb6M-buCgHhFOzTQxod27L9CqnOLio7N3gZAGpsp6n1-AJBOCJckcyXe2c6uD0srOJeZlKUm2eTDVMf8IehDVI0r1QOnTV6KzzAI3OY87Vd_cVMQ"
+      dockerfile: sts/Dockerfile
+    volumes:
+      - ./testfixtures_shared/shared:/fixture/shared
+    ports:
+      - "80"

test/fixtures/s3-fixture/src/main/java/fixture/s3/S3HttpFixtureWithSTS.java

@@ -0,0 +1,108 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+package fixture.s3;
+
+import com.sun.net.httpserver.HttpHandler;
+
+import org.elasticsearch.rest.RestStatus;
+
+import java.net.URLDecoder;
+import java.nio.charset.StandardCharsets;
+import java.time.ZonedDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.Arrays;
+import java.util.Map;
+import java.util.Objects;
+import java.util.stream.Collectors;
+
+public class S3HttpFixtureWithSTS extends S3HttpFixture {
+
+    private static final String ROLE_ARN = "arn:aws:iam::123456789012:role/FederatedWebIdentityRole";
+    private static final String ROLE_NAME = "sts-fixture-test";
+
+    private S3HttpFixtureWithSTS(final String[] args) throws Exception {
+        super(args);
+    }
+
+    @Override
+    protected HttpHandler createHandler(final String[] args) {
+        String accessKey = Objects.requireNonNull(args[4]);
+        String sessionToken = Objects.requireNonNull(args[5], "session token is missing");
+        String webIdentityToken = Objects.requireNonNull(args[6], "web identity token is missing");
+        final HttpHandler delegate = super.createHandler(args);
+
+        return exchange -> {
+            // https://docs.aws.amazon.com/STS/latest/APIReference/API_AssumeRoleWithWebIdentity.html
+            // It's run as a separate service, but we emulate it under the `assume-role-with-web-identity` endpoint
+            // of the S3 serve for the simplicity sake
+            if ("POST".equals(exchange.getRequestMethod())
+                && exchange.getRequestURI().getPath().startsWith("/assume-role-with-web-identity")) {
+                String body = new String(exchange.getRequestBody().readAllBytes(), StandardCharsets.UTF_8);
+                Map<String, String> params = Arrays.stream(body.split("&"))
+                    .map(e -> e.split("="))
+                    .collect(Collectors.toMap(e -> e[0], e -> URLDecoder.decode(e[1], StandardCharsets.UTF_8)));
+                if ("AssumeRoleWithWebIdentity".equals(params.get("Action")) == false) {
+                    exchange.sendResponseHeaders(RestStatus.BAD_REQUEST.getStatus(), 0);
+                    exchange.close();
+                    return;
+                }
+                if (ROLE_NAME.equals(params.get("RoleSessionName")) == false
+                    || webIdentityToken.equals(params.get("WebIdentityToken")) == false
+                    || ROLE_ARN.equals(params.get("RoleArn")) == false) {
+                    exchange.sendResponseHeaders(RestStatus.UNAUTHORIZED.getStatus(), 0);
+                    exchange.close();
+                    return;
+                }
+                final byte[] response = """
+                    <AssumeRoleWithWebIdentityResponse xmlns="https://sts.amazonaws.com/doc/2011-06-15/">
+                      <AssumeRoleWithWebIdentityResult>
+                        <SubjectFromWebIdentityToken>amzn1.account.AF6RHO7KZU5XRVQJGXK6HB56KR2A</SubjectFromWebIdentityToken>
+                        <Audience>client.5498841531868486423.1548@apps.example.com</Audience>
+                        <AssumedRoleUser>
+                          <Arn>%s</Arn>
+                          <AssumedRoleId>AROACLKWSDQRAOEXAMPLE:%s</AssumedRoleId>
+                        </AssumedRoleUser>
+                        <Credentials>
+                          <SessionToken>%s</SessionToken>
+                          <SecretAccessKey>secret_access_key</SecretAccessKey>
+                          <Expiration>%s</Expiration>
+                          <AccessKeyId>%s</AccessKeyId>
+                        </Credentials>
+                        <SourceIdentity>SourceIdentityValue</SourceIdentity>
+                        <Provider>www.amazon.com</Provider>
+                      </AssumeRoleWithWebIdentityResult>
+                      <ResponseMetadata>
+                        <RequestId>ad4156e9-bce1-11e2-82e6-6b6efEXAMPLE</RequestId>
+                      </ResponseMetadata>
+                    </AssumeRoleWithWebIdentityResponse>""".formatted(
+                    ROLE_ARN,
+                    ROLE_NAME,
+                    sessionToken,
+                    ZonedDateTime.now().plusDays(1L).format(DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ssZ")),
+                    accessKey
+                ).getBytes(StandardCharsets.UTF_8);
+                exchange.getResponseHeaders().add("Content-Type", "text/xml; charset=UTF-8");
+                exchange.sendResponseHeaders(RestStatus.OK.getStatus(), response.length);
+                exchange.getResponseBody().write(response);
+                exchange.close();
+                return;
+            }
+            delegate.handle(exchange);
+        };
+    }
+
+    public static void main(final String[] args) throws Exception {
+        if (args == null || args.length < 7) {
+            throw new IllegalArgumentException(
+                "S3HttpFixtureWithSTS expects 7 arguments [address, port, bucket, base path, sts access id, sts session token, web identity token]"
+            );
+        }
+        final S3HttpFixtureWithSTS fixture = new S3HttpFixtureWithSTS(args);
+        fixture.start();
+    }
+}

test/fixtures/s3-fixture/sts/Dockerfile

@@ -0,0 +1,25 @@
+FROM ubuntu:20.04
+
+RUN apt-get update -qqy
+RUN apt-get install -qqy openjdk-17-jre-headless
+
+ARG fixtureClass
+ARG port
+ARG bucket
+ARG basePath
+ARG accessKey
+ARG sessionToken
+ARG webIdentityToken
+
+ENV S3_FIXTURE_CLASS=${fixtureClass}
+ENV S3_FIXTURE_PORT=${port}
+ENV S3_FIXTURE_BUCKET=${bucket}
+ENV S3_FIXTURE_BASE_PATH=${basePath}
+ENV S3_FIXTURE_ACCESS_KEY=${accessKey}
+ENV S3_FIXTURE_SESSION_TOKEN=${sessionToken}
+ENV S3_WEB_IDENTITY_TOKEN=${webIdentityToken}
+
+ENTRYPOINT exec java -classpath "/fixture/shared/*" \
+    $S3_FIXTURE_CLASS 0.0.0.0 "$S3_FIXTURE_PORT" "$S3_FIXTURE_BUCKET" "$S3_FIXTURE_BASE_PATH" "$S3_FIXTURE_ACCESS_KEY" "$S3_FIXTURE_SESSION_TOKEN" "$S3_WEB_IDENTITY_TOKEN"
+
+EXPOSE $port