Support use of IRSA for repository-s3 plugin credentials (#3475)

* Support use of IRSA for repository-s3 plugin credentials

Signed-off-by: Andriy Redko <andriy.redko@aiven.io>

* Address code review comments

Signed-off-by: Andriy Redko <andriy.redko@aiven.io>

* Address code review comments

Signed-off-by: Andriy Redko <andriy.redko@aiven.io>

plugins/repository-s3/build.gradle

@@ -51,6 +51,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}"

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

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

plugins/repository-s3/src/main/java/org/opensearch/repositories/s3/AmazonS3Reference.java

@@ -32,17 +32,39 @@
 
 package org.opensearch.repositories.s3;
 
+import com.amazonaws.auth.AWSCredentialsProvider;
 import com.amazonaws.services.s3.AmazonS3;
 import com.amazonaws.services.s3.AmazonS3Client;
+
+import org.opensearch.common.Nullable;
 import org.opensearch.common.concurrent.RefCountedReleasable;
 
+import java.io.Closeable;
+import java.io.IOException;
+
 /**
  * Handles the shutdown of the wrapped {@link AmazonS3Client} using reference
  * counting.
  */
 public class AmazonS3Reference extends RefCountedReleasable<AmazonS3> {
-
     AmazonS3Reference(AmazonS3 client) {
-        super("AWS_S3_CLIENT", client, client::shutdown);
+        this(client, null);
+    }
+
+    AmazonS3Reference(AmazonS3WithCredentials client) {
+        this(client.client(), client.credentials());
+    }
+
+    AmazonS3Reference(AmazonS3 client, @Nullable AWSCredentialsProvider credentials) {
+        super("AWS_S3_CLIENT", client, () -> {
+            client.shutdown();
+            if (credentials instanceof Closeable) {
+                try {
+                    ((Closeable) credentials).close();
+                } catch (IOException e) {
+                    /* Do nothing here */
+                }
+            }
+        });
     }
 }

plugins/repository-s3/src/main/java/org/opensearch/repositories/s3/AmazonS3WithCredentials.java

@@ -0,0 +1,39 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.repositories.s3;
+
+import com.amazonaws.auth.AWSCredentialsProvider;
+import com.amazonaws.services.s3.AmazonS3;
+
+import org.opensearch.common.Nullable;
+
+/**
+ * The holder of the AmazonS3 and AWSCredentialsProvider
+ */
+final class AmazonS3WithCredentials {
+    private final AmazonS3 client;
+    private final AWSCredentialsProvider credentials;
+
+    private AmazonS3WithCredentials(final AmazonS3 client, @Nullable final AWSCredentialsProvider credentials) {
+        this.client = client;
+        this.credentials = credentials;
+    }
+
+    AmazonS3 client() {
+        return client;
+    }
+
+    AWSCredentialsProvider credentials() {
+        return credentials;
+    }
+
+    static AmazonS3WithCredentials create(final AmazonS3 client, @Nullable final AWSCredentialsProvider credentials) {
+        return new AmazonS3WithCredentials(client, credentials);
+    }
+}

plugins/repository-s3/src/main/java/org/opensearch/repositories/s3/S3ClientSettings.java

@@ -67,6 +67,29 @@ final class S3ClientSettings {
     /** Placeholder client name for normalizing client settings in the repository settings. */
     private static final String PLACEHOLDER_CLIENT = "placeholder";
 
+    // Properties to support using IAM Roles for Service Accounts (IRSA)
+
+    /** The identity token file for connecting to s3. */
+    static final Setting.AffixSetting<String> IDENTITY_TOKEN_FILE_SETTING = Setting.affixKeySetting(
+        PREFIX,
+        "identity_token_file",
+        key -> SecureSetting.simpleString(key, Property.NodeScope)
+    );
+
+    /** The role ARN (Amazon Resource Name) for connecting to s3. */
+    static final Setting.AffixSetting<SecureString> ROLE_ARN_SETTING = Setting.affixKeySetting(
+        PREFIX,
+        "role_arn",
+        key -> SecureSetting.secureString(key, null)
+    );
+
+    /** The role session name for connecting to s3. */
+    static final Setting.AffixSetting<SecureString> ROLE_SESSION_NAME_SETTING = Setting.affixKeySetting(
+        PREFIX,
+        "role_session_name",
+        key -> SecureSetting.secureString(key, null)
+    );
+
     /** The access key (ie login id) for connecting to s3. */
     static final Setting.AffixSetting<SecureString> ACCESS_KEY_SETTING = Setting.affixKeySetting(
         PREFIX,
@@ -189,6 +212,9 @@ final class S3ClientSettings {
     /** Credentials to authenticate with s3. */
     final S3BasicCredentials credentials;
 
+    /** Credentials to authenticate with s3 using IAM Roles for Service Accounts (IRSA). */
+    final IrsaCredentials irsaCredentials;
+
     /** The s3 endpoint the client should talk to, or empty string to use the default. */
     final String endpoint;
 
@@ -221,6 +247,7 @@ final class S3ClientSettings {
 
     private S3ClientSettings(
         S3BasicCredentials credentials,
+        IrsaCredentials irsaCredentials,
         String endpoint,
         Protocol protocol,
         int readTimeoutMillis,
@@ -233,6 +260,7 @@ final class S3ClientSettings {
         ProxySettings proxySettings
     ) {
         this.credentials = credentials;
+        this.irsaCredentials = irsaCredentials;
         this.endpoint = endpoint;
         this.protocol = protocol;
         this.readTimeoutMillis = readTimeoutMillis;
@@ -301,6 +329,7 @@ final class S3ClientSettings {
         validateInetAddressFor(newProxyHost);
         return new S3ClientSettings(
             newCredentials,
+            irsaCredentials,
             newEndpoint,
             newProtocol,
             newReadTimeoutMillis,
@@ -396,12 +425,27 @@ final class S3ClientSettings {
         }
     }
 
+    private static IrsaCredentials loadIrsaCredentials(Settings settings, String clientName) {
+        String identityTokenFile = getConfigValue(settings, clientName, IDENTITY_TOKEN_FILE_SETTING);
+        try (
+            SecureString roleArn = getConfigValue(settings, clientName, ROLE_ARN_SETTING);
+            SecureString roleSessionName = getConfigValue(settings, clientName, ROLE_SESSION_NAME_SETTING)
+        ) {
+            if (identityTokenFile.length() != 0 || roleArn.length() != 0 || roleSessionName.length() != 0) {
+                return new IrsaCredentials(identityTokenFile.toString(), roleArn.toString(), roleSessionName.toString());
+            }
+
+            return null;
+        }
+    }
+
     // pkg private for tests
     /** Parse settings for a single client. */
     static S3ClientSettings getClientSettings(final Settings settings, final String clientName) {
         final Protocol awsProtocol = getConfigValue(settings, clientName, PROTOCOL_SETTING);
         return new S3ClientSettings(
             S3ClientSettings.loadCredentials(settings, clientName),
+            S3ClientSettings.loadIrsaCredentials(settings, clientName),
             getConfigValue(settings, clientName, ENDPOINT_SETTING),
             awsProtocol,
             Math.toIntExact(getConfigValue(settings, clientName, READ_TIMEOUT_SETTING).millis()),
@@ -482,7 +526,8 @@ final class S3ClientSettings {
             && proxySettings.equals(that.proxySettings)
             && Objects.equals(disableChunkedEncoding, that.disableChunkedEncoding)
             && Objects.equals(region, that.region)
-            && Objects.equals(signerOverride, that.signerOverride);
+            && Objects.equals(signerOverride, that.signerOverride)
+            && Objects.equals(irsaCredentials, that.irsaCredentials);
     }
 
     @Override
@@ -512,4 +557,51 @@ final class S3ClientSettings {
         }
         return defaultValue;
     }
+
+    /**
+     * Class to store IAM Roles for Service Accounts (IRSA) credentials
+     * See please: https://docs.aws.amazon.com/eks/latest/userguide/iam-roles-for-service-accounts.html
+     */
+    static class IrsaCredentials {
+        private final String identityTokenFile;
+        private final String roleArn;
+        private final String roleSessionName;
+
+        IrsaCredentials(String identityTokenFile, String roleArn, String roleSessionName) {
+            this.identityTokenFile = Strings.isNullOrEmpty(identityTokenFile) ? null : identityTokenFile;
+            this.roleArn = Strings.isNullOrEmpty(roleArn) ? null : roleArn;
+            this.roleSessionName = Strings.isNullOrEmpty(roleSessionName) ? "s3-sdk-java-" + System.currentTimeMillis() : roleSessionName;
+        }
+
+        public String getIdentityTokenFile() {
+            return identityTokenFile;
+        }
+
+        public String getRoleArn() {
+            return roleArn;
+        }
+
+        public String getRoleSessionName() {
+            return roleSessionName;
+        }
+
+        @Override
+        public boolean equals(final Object o) {
+            if (this == o) {
+                return true;
+            }
+            if (o == null || getClass() != o.getClass()) {
+                return false;
+            }
+            final IrsaCredentials that = (IrsaCredentials) o;
+            return Objects.equals(identityTokenFile, that.identityTokenFile)
+                && Objects.equals(roleArn, that.roleArn)
+                && Objects.equals(roleSessionName, that.roleSessionName);
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(identityTokenFile, roleArn, roleSessionName);
+        }
+    }
 }

plugins/repository-s3/src/main/java/org/opensearch/repositories/s3/S3RepositoryPlugin.java

@@ -132,7 +132,10 @@ public class S3RepositoryPlugin extends Plugin implements RepositoryPlugin, Relo
             S3Repository.ACCESS_KEY_SETTING,
             S3Repository.SECRET_KEY_SETTING,
             S3ClientSettings.SIGNER_OVERRIDE,
-            S3ClientSettings.REGION
+            S3ClientSettings.REGION,
+            S3ClientSettings.ROLE_ARN_SETTING,
+            S3ClientSettings.IDENTITY_TOKEN_FILE_SETTING,
+            S3ClientSettings.ROLE_SESSION_NAME_SETTING
         );
     }
 

plugins/repository-s3/src/main/java/org/opensearch/repositories/s3/S3Service.java

@@ -35,8 +35,11 @@ package org.opensearch.repositories.s3;
 import com.amazonaws.ClientConfiguration;
 import com.amazonaws.auth.AWSCredentials;
 import com.amazonaws.auth.AWSCredentialsProvider;
+import com.amazonaws.auth.AWSSessionCredentialsProvider;
 import com.amazonaws.auth.AWSStaticCredentialsProvider;
 import com.amazonaws.auth.EC2ContainerCredentialsProviderWrapper;
+import com.amazonaws.auth.STSAssumeRoleSessionCredentialsProvider;
+import com.amazonaws.auth.STSAssumeRoleWithWebIdentitySessionCredentialsProvider;
 import com.amazonaws.client.builder.AwsClientBuilder;
 import com.amazonaws.http.IdleConnectionReaper;
 import com.amazonaws.http.SystemPropertyTlsKeyManagersProvider;
@@ -45,6 +48,8 @@ import com.amazonaws.internal.SdkSSLContext;
 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.AWSSecurityTokenServiceClientBuilder;
 
 import org.apache.http.conn.ssl.DefaultHostnameVerifier;
 import org.apache.http.conn.ssl.SSLConnectionSocketFactory;
@@ -52,9 +57,11 @@ import org.apache.http.protocol.HttpContext;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.opensearch.cluster.metadata.RepositoryMetadata;
+import org.opensearch.common.Nullable;
 import org.opensearch.common.Strings;
 import org.opensearch.common.collect.MapBuilder;
 import org.opensearch.common.settings.Settings;
+import org.opensearch.repositories.s3.S3ClientSettings.IrsaCredentials;
 
 import javax.net.ssl.SSLContext;
 import java.io.Closeable;
@@ -67,6 +74,9 @@ import java.net.Socket;
 import java.security.SecureRandom;
 import java.util.Map;
 
+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 {
@@ -163,9 +173,11 @@ class S3Service implements Closeable {
     }
 
     // proxy for testing
-    AmazonS3 buildClient(final S3ClientSettings clientSettings) {
+    AmazonS3WithCredentials buildClient(final S3ClientSettings clientSettings) {
         final AmazonS3ClientBuilder builder = AmazonS3ClientBuilder.standard();
-        builder.withCredentials(buildCredentials(logger, clientSettings));
+
+        final AWSCredentialsProvider credentials = buildCredentials(logger, clientSettings);
+        builder.withCredentials(credentials);
         builder.withClientConfiguration(buildConfiguration(clientSettings));
 
         String endpoint = Strings.hasLength(clientSettings.endpoint) ? clientSettings.endpoint : Constants.S3_HOSTNAME;
@@ -192,7 +204,8 @@ class S3Service implements Closeable {
         if (clientSettings.disableChunkedEncoding) {
             builder.disableChunkedEncoding();
         }
-        return SocketAccess.doPrivileged(builder::build);
+        final AmazonS3 client = SocketAccess.doPrivileged(builder::build);
+        return AmazonS3WithCredentials.create(client, credentials);
     }
 
     // pkg private for tests
@@ -258,24 +271,83 @@ class S3Service implements Closeable {
 
     // pkg private for tests
     static AWSCredentialsProvider buildCredentials(Logger logger, S3ClientSettings clientSettings) {
-        final S3BasicCredentials credentials = clientSettings.credentials;
+        final S3BasicCredentials basicCredentials = clientSettings.credentials;
-        if (credentials == null) {
+        final IrsaCredentials irsaCredentials = buildFromEnviroment(clientSettings.irsaCredentials);
+
+        // If IAM Roles for Service Accounts (IRSA) credentials are configured, start with them first
+        if (irsaCredentials != null) {
+            logger.debug("Using IRSA credentials");
+
+            AWSSecurityTokenService securityTokenService = null;
+            final String region = Strings.hasLength(clientSettings.region) ? clientSettings.region : null;
+            if (region != null || basicCredentials != null) {
+                securityTokenService = SocketAccess.doPrivileged(
+                    () -> AWSSecurityTokenServiceClientBuilder.standard()
+                        .withCredentials((basicCredentials != null) ? new AWSStaticCredentialsProvider(basicCredentials) : null)
+                        .withRegion(region)
+                        .build()
+                );
+            }
+
+            if (irsaCredentials.getIdentityTokenFile() == null) {
+                return new PrivilegedSTSAssumeRoleSessionCredentialsProvider<>(
+                    securityTokenService,
+                    new STSAssumeRoleSessionCredentialsProvider.Builder(irsaCredentials.getRoleArn(), irsaCredentials.getRoleSessionName())
+                        .withStsClient(securityTokenService)
+                        .build()
+                );
+            } else {
+                return new PrivilegedSTSAssumeRoleSessionCredentialsProvider<>(
+                    securityTokenService,
+                    new STSAssumeRoleWithWebIdentitySessionCredentialsProvider.Builder(
+                        irsaCredentials.getRoleArn(),
+                        irsaCredentials.getRoleSessionName(),
+                        irsaCredentials.getIdentityTokenFile()
+                    ).withStsClient(securityTokenService).build()
+                );
+            }
+        } else if (basicCredentials != null) {
+            logger.debug("Using basic key/secret credentials");
+            return new AWSStaticCredentialsProvider(basicCredentials);
+        } else {
             logger.debug("Using instance profile credentials");
             return new PrivilegedInstanceProfileCredentialsProvider();
-        } else {
-            logger.debug("Using basic key/secret credentials");
-            return new AWSStaticCredentialsProvider(credentials);
         }
     }
 
+    private static IrsaCredentials buildFromEnviroment(IrsaCredentials defaults) {
+        if (defaults == null) {
+            return null;
+        }
+
+        String webIdentityTokenFile = defaults.getIdentityTokenFile();
+        if (webIdentityTokenFile == null) {
+            webIdentityTokenFile = System.getenv(AWS_WEB_IDENTITY_ENV_VAR);
+        }
+
+        String roleArn = defaults.getRoleArn();
+        if (roleArn == null) {
+            roleArn = System.getenv(AWS_ROLE_ARN_ENV_VAR);
+        }
+
+        String roleSessionName = defaults.getRoleSessionName();
+        if (roleSessionName == null) {
+            roleSessionName = System.getenv(AWS_ROLE_SESSION_NAME_ENV_VAR);
+        }
+
+        return new IrsaCredentials(webIdentityTokenFile, roleArn, roleSessionName);
+    }
+
     private synchronized void releaseCachedClients() {
         // the clients will shutdown when they will not be used anymore
         for (final AmazonS3Reference clientReference : clientsCache.values()) {
             clientReference.decRef();
         }
+
         // clear previously cached clients, they will be build lazily
         clientsCache = emptyMap();
         derivedClientSettings = emptyMap();
+
         // shutdown IdleConnectionReaper background thread
         // it will be restarted on new client usage
         IdleConnectionReaper.shutdown();
@@ -300,6 +372,43 @@ class S3Service implements Closeable {
         }
     }
 
+    static class PrivilegedSTSAssumeRoleSessionCredentialsProvider<P extends AWSSessionCredentialsProvider & Closeable>
+        implements
+            AWSCredentialsProvider,
+            Closeable {
+        private final P credentials;
+        private final AWSSecurityTokenService securityTokenService;
+
+        private PrivilegedSTSAssumeRoleSessionCredentialsProvider(
+            @Nullable final AWSSecurityTokenService securityTokenService,
+            final P credentials
+        ) {
+            this.securityTokenService = securityTokenService;
+            this.credentials = credentials;
+        }
+
+        @Override
+        public AWSCredentials getCredentials() {
+            return SocketAccess.doPrivileged(credentials::getCredentials);
+        }
+
+        @Override
+        public void refresh() {
+            SocketAccess.doPrivilegedVoid(credentials::refresh);
+        }
+
+        @Override
+        public void close() throws IOException {
+            SocketAccess.doPrivilegedIOException(() -> {
+                credentials.close();
+                if (securityTokenService != null) {
+                    securityTokenService.shutdown();
+                }
+                return null;
+            });
+        };
+    }
+
     @Override
     public void close() {
         releaseCachedClients();

plugins/repository-s3/src/test/java/org/opensearch/repositories/s3/AwsS3ServiceImplTests.java

@@ -36,11 +36,16 @@ import com.amazonaws.ClientConfiguration;
 import com.amazonaws.Protocol;
 import com.amazonaws.auth.AWSCredentialsProvider;
 import com.amazonaws.auth.AWSStaticCredentialsProvider;
+import com.amazonaws.http.IdleConnectionReaper;
+
+import org.junit.AfterClass;
 import org.opensearch.common.settings.MockSecureSettings;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.test.OpenSearchTestCase;
 
+import java.io.Closeable;
 import java.io.IOException;
+import java.util.HashMap;
 import java.util.Locale;
 import java.util.Map;
 
@@ -51,6 +56,11 @@ import static org.opensearch.repositories.s3.S3ClientSettings.PROTOCOL_SETTING;
 import static org.opensearch.repositories.s3.S3ClientSettings.PROXY_TYPE_SETTING;
 
 public class AwsS3ServiceImplTests extends OpenSearchTestCase {
+    @AfterClass
+    public static void shutdownIdleConnectionReaper() {
+        // created by default STS client
+        IdleConnectionReaper.shutdown();
+    }
 
     public void testAWSCredentialsDefaultToInstanceProviders() {
         final String inexistentClientName = randomAlphaOfLength(8).toLowerCase(Locale.ROOT);
@@ -86,6 +96,101 @@ public class AwsS3ServiceImplTests extends OpenSearchTestCase {
         assertThat(defaultCredentialsProvider, instanceOf(S3Service.PrivilegedInstanceProfileCredentialsProvider.class));
     }
 
+    public void testCredentialsAndIrsaWithIdentityTokenFileCredentialsFromKeystore() throws IOException {
+        final Map<String, String> plainSettings = new HashMap<>();
+        final MockSecureSettings secureSettings = new MockSecureSettings();
+        final String clientNamePrefix = "some_client_name_";
+        final int clientsCount = randomIntBetween(0, 4);
+        for (int i = 0; i < clientsCount; i++) {
+            final String clientName = clientNamePrefix + i;
+            secureSettings.setString("s3.client." + clientName + ".role_arn", clientName + "_role_arn");
+
+            // Use static AWS credentials for tests
+            secureSettings.setString("s3.client." + clientName + ".access_key", clientName + "_aws_access_key");
+            secureSettings.setString("s3.client." + clientName + ".secret_key", clientName + "_aws_secret_key");
+
+            // Use explicit region setting
+            plainSettings.put("s3.client." + clientName + ".region", "us-east1");
+            plainSettings.put("s3.client." + clientName + ".identity_token_file", clientName + "_identity_token_file");
+        }
+        final Settings settings = Settings.builder().loadFromMap(plainSettings).setSecureSettings(secureSettings).build();
+        final Map<String, S3ClientSettings> allClientsSettings = S3ClientSettings.load(settings);
+        // no less, no more
+        assertThat(allClientsSettings.size(), is(clientsCount + 1)); // including default
+        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);
+            assertThat(credentialsProvider, instanceOf(S3Service.PrivilegedSTSAssumeRoleSessionCredentialsProvider.class));
+            ((Closeable) credentialsProvider).close();
+        }
+        // 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));
+    }
+
+    public void testCredentialsAndIrsaCredentialsFromKeystore() throws IOException {
+        final Map<String, String> plainSettings = new HashMap<>();
+        final MockSecureSettings secureSettings = new MockSecureSettings();
+        final String clientNamePrefix = "some_client_name_";
+        final int clientsCount = randomIntBetween(0, 4);
+        for (int i = 0; i < clientsCount; i++) {
+            final String clientName = clientNamePrefix + i;
+            secureSettings.setString("s3.client." + clientName + ".role_arn", clientName + "_role_arn");
+            secureSettings.setString("s3.client." + clientName + ".role_session_name", clientName + "_role_session_name");
+
+            // Use static AWS credentials for tests
+            secureSettings.setString("s3.client." + clientName + ".access_key", clientName + "_aws_access_key");
+            secureSettings.setString("s3.client." + clientName + ".secret_key", clientName + "_aws_secret_key");
+
+            // Use explicit region setting
+            plainSettings.put("s3.client." + clientName + ".region", "us-east1");
+        }
+        final Settings settings = Settings.builder().loadFromMap(plainSettings).setSecureSettings(secureSettings).build();
+        final Map<String, S3ClientSettings> allClientsSettings = S3ClientSettings.load(settings);
+        // no less, no more
+        assertThat(allClientsSettings.size(), is(clientsCount + 1)); // including default
+        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);
+            assertThat(credentialsProvider, instanceOf(S3Service.PrivilegedSTSAssumeRoleSessionCredentialsProvider.class));
+            ((Closeable) credentialsProvider).close();
+        }
+        // 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));
+    }
+
+    public void testIrsaCredentialsFromKeystore() throws IOException {
+        final Map<String, String> plainSettings = new HashMap<>();
+        final MockSecureSettings secureSettings = new MockSecureSettings();
+        final String clientNamePrefix = "some_client_name_";
+        final int clientsCount = randomIntBetween(0, 4);
+        for (int i = 0; i < clientsCount; i++) {
+            final String clientName = clientNamePrefix + i;
+            secureSettings.setString("s3.client." + clientName + ".role_arn", clientName + "_role_arn");
+            secureSettings.setString("s3.client." + clientName + ".role_session_name", clientName + "_role_session_name");
+        }
+        final Settings settings = Settings.builder().loadFromMap(plainSettings).setSecureSettings(secureSettings).build();
+        final Map<String, S3ClientSettings> allClientsSettings = S3ClientSettings.load(settings);
+        // no less, no more
+        assertThat(allClientsSettings.size(), is(clientsCount + 1)); // including default
+        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);
+            assertThat(credentialsProvider, instanceOf(S3Service.PrivilegedSTSAssumeRoleSessionCredentialsProvider.class));
+            ((Closeable) credentialsProvider).close();
+        }
+        // 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));
+    }
+
     public void testSetDefaultCredential() {
         final MockSecureSettings secureSettings = new MockSecureSettings();
         final String awsAccessKey = randomAlphaOfLength(8);

plugins/repository-s3/src/test/java/org/opensearch/repositories/s3/RepositoryCredentialsTests.java

@@ -317,9 +317,10 @@ public class RepositoryCredentialsTests extends OpenSearchSingleNodeTestCase {
             private static final Logger logger = LogManager.getLogger(ProxyS3Service.class);
 
             @Override
-            AmazonS3 buildClient(final S3ClientSettings clientSettings) {
+            AmazonS3WithCredentials buildClient(final S3ClientSettings clientSettings) {
-                final AmazonS3 client = super.buildClient(clientSettings);
+                final AmazonS3WithCredentials client = super.buildClient(clientSettings);
-                return new ClientAndCredentials(client, buildCredentials(logger, clientSettings));
+                final AWSCredentialsProvider credentials = buildCredentials(logger, clientSettings);
+                return AmazonS3WithCredentials.create(new ClientAndCredentials(client.client(), credentials), credentials);
             }
 
         }

plugins/repository-s3/src/test/java/org/opensearch/repositories/s3/S3ClientSettingsTests.java

@@ -45,6 +45,7 @@ import java.net.InetSocketAddress;
 import java.util.Locale;
 import java.util.Map;
 
+import static org.hamcrest.CoreMatchers.startsWith;
 import static org.hamcrest.Matchers.contains;
 import static org.hamcrest.Matchers.emptyString;
 import static org.hamcrest.Matchers.is;
@@ -118,6 +119,52 @@ public class S3ClientSettingsTests extends OpenSearchTestCase {
         assertThat(e.getMessage(), is("Missing access key and secret key for s3 client [default]"));
     }
 
+    public void testIrsaCredentialsTypeWithIdentityTokenFile() {
+        final Map<String, S3ClientSettings> settings = S3ClientSettings.load(
+            Settings.builder().put("s3.client.default.identity_token_file", "file").build()
+        );
+        final S3ClientSettings defaultSettings = settings.get("default");
+        final S3ClientSettings.IrsaCredentials credentials = defaultSettings.irsaCredentials;
+        assertThat(credentials.getIdentityTokenFile(), is("file"));
+        assertThat(credentials.getRoleArn(), is(nullValue()));
+        assertThat(credentials.getRoleSessionName(), startsWith("s3-sdk-java-"));
+    }
+
+    public void testIrsaCredentialsTypeRoleArn() {
+        final MockSecureSettings secureSettings = new MockSecureSettings();
+        secureSettings.setString("s3.client.default.role_arn", "role");
+        final Map<String, S3ClientSettings> settings = S3ClientSettings.load(Settings.builder().setSecureSettings(secureSettings).build());
+        final S3ClientSettings defaultSettings = settings.get("default");
+        final S3ClientSettings.IrsaCredentials credentials = defaultSettings.irsaCredentials;
+        assertThat(credentials.getRoleArn(), is("role"));
+        assertThat(credentials.getRoleSessionName(), startsWith("s3-sdk-java-"));
+    }
+
+    public void testIrsaCredentialsTypeWithRoleArnAndRoleSessionName() {
+        final MockSecureSettings secureSettings = new MockSecureSettings();
+        secureSettings.setString("s3.client.default.role_arn", "role");
+        secureSettings.setString("s3.client.default.role_session_name", "session");
+        final Map<String, S3ClientSettings> settings = S3ClientSettings.load(Settings.builder().setSecureSettings(secureSettings).build());
+        final S3ClientSettings defaultSettings = settings.get("default");
+        final S3ClientSettings.IrsaCredentials credentials = defaultSettings.irsaCredentials;
+        assertThat(credentials.getRoleArn(), is("role"));
+        assertThat(credentials.getRoleSessionName(), is("session"));
+    }
+
+    public void testIrsaCredentialsTypeWithRoleArnAndRoleSessionNameAndIdentityTokeFile() {
+        final MockSecureSettings secureSettings = new MockSecureSettings();
+        secureSettings.setString("s3.client.default.role_arn", "role");
+        secureSettings.setString("s3.client.default.role_session_name", "session");
+        final Map<String, S3ClientSettings> settings = S3ClientSettings.load(
+            Settings.builder().setSecureSettings(secureSettings).put("s3.client.default.identity_token_file", "file").build()
+        );
+        final S3ClientSettings defaultSettings = settings.get("default");
+        final S3ClientSettings.IrsaCredentials credentials = defaultSettings.irsaCredentials;
+        assertThat(credentials.getIdentityTokenFile(), is("file"));
+        assertThat(credentials.getRoleArn(), is("role"));
+        assertThat(credentials.getRoleSessionName(), is("session"));
+    }
+
     public void testCredentialsTypeWithAccessKeyAndSecretKey() {
         final MockSecureSettings secureSettings = new MockSecureSettings();
         secureSettings.setString("s3.client.default.access_key", "access_key");
@@ -199,7 +246,7 @@ public class S3ClientSettingsTests extends OpenSearchTestCase {
         assertThat(settings.get("default").region, is(""));
         assertThat(settings.get("other").region, is(region));
         try (S3Service s3Service = new S3Service()) {
-            AmazonS3Client other = (AmazonS3Client) s3Service.buildClient(settings.get("other"));
+            AmazonS3Client other = (AmazonS3Client) s3Service.buildClient(settings.get("other")).client();
             assertThat(other.getSignerRegionOverride(), is(region));
         }
     }

plugins/repository-s3/src/test/java/org/opensearch/repositories/s3/S3ServiceTests.java

@@ -32,10 +32,12 @@
 package org.opensearch.repositories.s3;
 
 import org.opensearch.cluster.metadata.RepositoryMetadata;
-
+import org.opensearch.common.settings.MockSecureSettings;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.test.OpenSearchTestCase;
 
+import java.util.Map;
+
 public class S3ServiceTests extends OpenSearchTestCase {
 
     public void testCachedClientsAreReleased() {
@@ -56,4 +58,29 @@ public class S3ServiceTests extends OpenSearchTestCase {
         final S3ClientSettings clientSettingsReloaded = s3Service.settings(metadata1);
         assertNotSame(clientSettings, clientSettingsReloaded);
     }
+
+    public void testCachedClientsWithCredentialsAreReleased() {
+        final MockSecureSettings secureSettings = new MockSecureSettings();
+        secureSettings.setString("s3.client.default.role_arn", "role");
+        final Map<String, S3ClientSettings> defaults = S3ClientSettings.load(
+            Settings.builder().setSecureSettings(secureSettings).put("s3.client.default.identity_token_file", "file").build()
+        );
+        final S3Service s3Service = new S3Service();
+        s3Service.refreshAndClearCache(defaults);
+        final Settings settings = Settings.builder().put("endpoint", "http://first").put("region", "us-east-2").build();
+        final RepositoryMetadata metadata1 = new RepositoryMetadata("first", "s3", settings);
+        final RepositoryMetadata metadata2 = new RepositoryMetadata("second", "s3", settings);
+        final S3ClientSettings clientSettings = s3Service.settings(metadata2);
+        final S3ClientSettings otherClientSettings = s3Service.settings(metadata2);
+        assertSame(clientSettings, otherClientSettings);
+        final AmazonS3Reference reference = s3Service.client(metadata1);
+        reference.close();
+        s3Service.close();
+        final AmazonS3Reference referenceReloaded = s3Service.client(metadata1);
+        assertNotSame(referenceReloaded, reference);
+        referenceReloaded.close();
+        s3Service.close();
+        final S3ClientSettings clientSettingsReloaded = s3Service.settings(metadata1);
+        assertNotSame(clientSettings, clientSettingsReloaded);
+    }
 }