Add support for AWS session tokens

AWS supports the creation and use of credentials that are only valid for a
fixed period of time. These credentials comprise three parts: the usual access
key and secret key, together with a session token. This commit adds support for
these three-part credentials to the EC2 discovery plugin and the S3 repository
plugin.

Note that session tokens are only valid for a limited period of time and yet
there is no mechanism for refreshing or rotating them when they expire without
restarting Elasticsearch.  Nonetheless, this feature is already useful for
nodes that need only run for a few days, such as for training, testing or
evaluation. elastic#29135 tracks the work towards allowing these credentials to be
refreshed at runtime.

Resolves elastic#16428

docs/CHANGELOG.asciidoc

@@ -149,6 +149,9 @@ option. ({pull}30140[#29658])
 A new analysis plugin called `analysis_nori` that exposes the Lucene Korean
 analysis module.  ({pull}30397[#30397])
 
+The `discovery-ec2` and `repository-s3` plugins now have support for AWS
+session credentials. ({pull}TODO[#TODO])
+
 [float]
 === Enhancements
 

docs/plugins/discovery-ec2.asciidoc

@@ -46,6 +46,11 @@ Those that must be stored in the keystore are marked as `Secure`.
 
     An s3 secret key. The `access_key` setting must also be specified. (Secure)
 
+`session_token`::
+
+    An s3 session token. The `access_key` and `secret_key` settings must also
+    be specified. (Secure)
+
 `endpoint`::
 
     The ec2 service endpoint to connect to. This will be automatically

docs/plugins/repository-s3.asciidoc

@@ -73,6 +73,11 @@ are marked as `Secure`.
 
     An s3 secret key. The `access_key` setting must also be specified. (Secure)
 
+`session_token`::
+
+    An s3 session token. The `access_key` and `secret_key` settings must also
+    be specified. (Secure)
+
 `endpoint`::
 
     The s3 service endpoint to connect to. This will be automatically

plugins/discovery-ec2/src/main/java/org/elasticsearch/discovery/ec2/AwsEc2Service.java

@@ -52,6 +52,9 @@ class HostType {
     /** The secret key (ie password) for connecting to ec2. */
     Setting<SecureString> SECRET_KEY_SETTING = SecureSetting.secureString("discovery.ec2.secret_key", null);
 
+    /** The (optional) session token for connecting to ec2. */
+    Setting<SecureString> SESSION_TOKEN_SETTING = SecureSetting.secureString("discovery.ec2.session_token", null);
+
     /** An override for the ec2 endpoint to connect to. */
     Setting<String> ENDPOINT_SETTING = new Setting<>("discovery.ec2.endpoint", "",
         s -> s.toLowerCase(Locale.ROOT), Property.NodeScope);

plugins/discovery-ec2/src/main/java/org/elasticsearch/discovery/ec2/AwsEc2ServiceImpl.java

@@ -26,22 +26,30 @@
 import com.amazonaws.AmazonClientException;
 import com.amazonaws.AmazonWebServiceRequest;
 import com.amazonaws.ClientConfiguration;
+import com.amazonaws.auth.AWSCredentials;
 import com.amazonaws.auth.AWSCredentialsProvider;
+import com.amazonaws.auth.AWSStaticCredentialsProvider;
 import com.amazonaws.auth.BasicAWSCredentials;
+import com.amazonaws.auth.BasicSessionCredentials;
 import com.amazonaws.auth.DefaultAWSCredentialsProviderChain;
 import com.amazonaws.http.IdleConnectionReaper;
 import com.amazonaws.internal.StaticCredentialsProvider;
 import com.amazonaws.retry.RetryPolicy;
 import com.amazonaws.services.ec2.AmazonEC2;
 import com.amazonaws.services.ec2.AmazonEC2Client;
 import org.apache.logging.log4j.Logger;
+import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.common.Randomness;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.component.AbstractComponent;
+import org.elasticsearch.common.logging.DeprecationLogger;
+import org.elasticsearch.common.logging.Loggers;
 import org.elasticsearch.common.settings.SecureString;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.settings.SettingsException;
 
 class AwsEc2ServiceImpl extends AbstractComponent implements AwsEc2Service, Closeable {
+    private static final DeprecationLogger DEPRECATION_LOGGER = new DeprecationLogger(Loggers.getLogger(AwsEc2ServiceImpl.class));
 
     public static final String EC2_METADATA_URL = "http://169.254.169.254/latest/meta-data/";
 
@@ -67,20 +75,38 @@ public synchronized AmazonEC2 client() {
     }
 
     protected static AWSCredentialsProvider buildCredentials(Logger logger, Settings settings) {
-        AWSCredentialsProvider credentials;
-
         try (SecureString key = ACCESS_KEY_SETTING.get(settings);
-             SecureString secret = SECRET_KEY_SETTING.get(settings)) {
+             SecureString secret = SECRET_KEY_SETTING.get(settings);
+             SecureString sessionToken = SESSION_TOKEN_SETTING.get(settings)) {
             if (key.length() == 0 && secret.length() == 0) {
+                if (sessionToken.length() > 0) {
+                    throw new SettingsException("Setting [{}] is set but [{}] and [{}] are not",
+                        SESSION_TOKEN_SETTING.getKey(), ACCESS_KEY_SETTING.getKey(), SECRET_KEY_SETTING.getKey());
+                }
+
                 logger.debug("Using either environment variables, system properties or instance profile credentials");
-                credentials = new DefaultAWSCredentialsProviderChain();
+                return new DefaultAWSCredentialsProviderChain();
             } else {
-                logger.debug("Using basic key/secret credentials");
-                credentials = new StaticCredentialsProvider(new BasicAWSCredentials(key.toString(), secret.toString()));
+                if (key.length() == 0) {
+                    DEPRECATION_LOGGER.deprecated("Setting [{}] is set but [{}] is not, which will be unsupported in future",
+                        SECRET_KEY_SETTING.getKey(), ACCESS_KEY_SETTING.getKey());
+                }
+                if (secret.length() == 0) {
+                    DEPRECATION_LOGGER.deprecated("Setting [{}] is set but [{}] is not, which will be unsupported in future",
+                        ACCESS_KEY_SETTING.getKey(), SECRET_KEY_SETTING.getKey());
+                }
+
+                final AWSCredentials credentials;
+                if (sessionToken.length() == 0) {
+                    logger.debug("Using basic key/secret credentials");
+                    credentials = new BasicAWSCredentials(key.toString(), secret.toString());
+                } else {
+                    logger.debug("Using basic session credentials");
+                    credentials = new BasicSessionCredentials(key.toString(), secret.toString(), sessionToken.toString());
+                }
+                return new AWSStaticCredentialsProvider(credentials);
             }
         }
-
-        return credentials;
     }
 
     protected static ClientConfiguration buildConfiguration(Logger logger, Settings settings) {

plugins/discovery-ec2/src/main/java/org/elasticsearch/discovery/ec2/Ec2DiscoveryPlugin.java

@@ -106,6 +106,7 @@ public List<Setting<?>> getSettings() {
         // Register EC2 discovery settings: discovery.ec2
         AwsEc2Service.ACCESS_KEY_SETTING,
         AwsEc2Service.SECRET_KEY_SETTING,
+        AwsEc2Service.SESSION_TOKEN_SETTING,
         AwsEc2Service.ENDPOINT_SETTING,
         AwsEc2Service.PROTOCOL_SETTING,
         AwsEc2Service.PROXY_HOST_SETTING,

plugins/discovery-ec2/src/test/java/org/elasticsearch/discovery/ec2/AwsEc2ServiceImplTests.java

@@ -23,15 +23,13 @@
 import com.amazonaws.Protocol;
 import com.amazonaws.auth.AWSCredentials;
 import com.amazonaws.auth.AWSCredentialsProvider;
+import com.amazonaws.auth.BasicSessionCredentials;
 import com.amazonaws.auth.DefaultAWSCredentialsProviderChain;
 import org.elasticsearch.common.settings.MockSecureSettings;
-import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.discovery.ec2.AwsEc2Service;
-import org.elasticsearch.discovery.ec2.AwsEc2ServiceImpl;
+import org.elasticsearch.common.settings.SettingsException;
 import org.elasticsearch.test.ESTestCase;
 
-import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.nullValue;
@@ -44,17 +42,56 @@ public void testAWSCredentialsWithSystemProviders() {
     }
 
     public void testAWSCredentialsWithElasticsearchAwsSettings() {
-        MockSecureSettings secureSettings = new MockSecureSettings();
+        final MockSecureSettings secureSettings = new MockSecureSettings();
+        secureSettings.setString("discovery.ec2.access_key", "aws_key");
+        secureSettings.setString("discovery.ec2.secret_key", "aws_secret");
+        final AWSCredentials credentials = AwsEc2ServiceImpl.buildCredentials(logger,
+            Settings.builder().setSecureSettings(secureSettings).build()).getCredentials();
+        assertThat(credentials.getAWSAccessKeyId(), is("aws_key"));
+        assertThat(credentials.getAWSSecretKey(), is("aws_secret"));
+    }
+
+    public void testAWSSessionCredentialsWithElasticsearchAwsSettings() {
+        final MockSecureSettings secureSettings = new MockSecureSettings();
         secureSettings.setString("discovery.ec2.access_key", "aws_key");
         secureSettings.setString("discovery.ec2.secret_key", "aws_secret");
-        Settings settings = Settings.builder().setSecureSettings(secureSettings).build();
-        launchAWSCredentialsWithElasticsearchSettingsTest(settings, "aws_key", "aws_secret");
+        secureSettings.setString("discovery.ec2.session_token", "aws_session_token");
+        final BasicSessionCredentials credentials = (BasicSessionCredentials) AwsEc2ServiceImpl.buildCredentials(logger,
+            Settings.builder().setSecureSettings(secureSettings).build()).getCredentials();
+        assertThat(credentials.getAWSAccessKeyId(), is("aws_key"));
+        assertThat(credentials.getAWSSecretKey(), is("aws_secret"));
+        assertThat(credentials.getSessionToken(), is("aws_session_token"));
+    }
+
+    public void testDeprecationOfLoneAccessKey() {
+        final MockSecureSettings secureSettings = new MockSecureSettings();
+        secureSettings.setString("discovery.ec2.access_key", "aws_key");
+        final AWSCredentials credentials = AwsEc2ServiceImpl.buildCredentials(logger,
+            Settings.builder().setSecureSettings(secureSettings).build()).getCredentials();
+        assertThat(credentials.getAWSAccessKeyId(), is("aws_key"));
+        assertThat(credentials.getAWSSecretKey(), is(""));
+        assertSettingDeprecationsAndWarnings(new String[]{},
+            "Setting [discovery.ec2.access_key] is set but [discovery.ec2.secret_key] is not, which will be unsupported in future");
+    }
+
+    public void testDeprecationOfLoneSecretKey() {
+        final MockSecureSettings secureSettings = new MockSecureSettings();
+        secureSettings.setString("discovery.ec2.secret_key", "aws_secret");
+        final AWSCredentials credentials = AwsEc2ServiceImpl.buildCredentials(logger,
+            Settings.builder().setSecureSettings(secureSettings).build()).getCredentials();
+        assertThat(credentials.getAWSAccessKeyId(), is(""));
+        assertThat(credentials.getAWSSecretKey(), is("aws_secret"));
+        assertSettingDeprecationsAndWarnings(new String[]{},
+            "Setting [discovery.ec2.secret_key] is set but [discovery.ec2.access_key] is not, which will be unsupported in future");
     }
 
-    protected void launchAWSCredentialsWithElasticsearchSettingsTest(Settings settings, String expectedKey, String expectedSecret) {
-        AWSCredentials credentials = AwsEc2ServiceImpl.buildCredentials(logger, settings).getCredentials();
-        assertThat(credentials.getAWSAccessKeyId(), is(expectedKey));
-        assertThat(credentials.getAWSSecretKey(), is(expectedSecret));
+    public void testRejectionOfLoneSessionToken() {
+        final MockSecureSettings secureSettings = new MockSecureSettings();
+        secureSettings.setString("discovery.ec2.session_token", "aws_session_token");
+        SettingsException e = expectThrows(SettingsException.class, () -> AwsEc2ServiceImpl.buildCredentials(logger,
+            Settings.builder().setSecureSettings(secureSettings).build()));
+        assertThat(e.getMessage(), is(
+            "Setting [discovery.ec2.session_token] is set but [discovery.ec2.access_key] and [discovery.ec2.secret_key] are not"));
     }
 
     public void testAWSDefaultConfiguration() {

plugins/discovery-ec2/src/test/java/org/elasticsearch/discovery/ec2/Ec2DiscoveryClusterFormationTests.java

@@ -82,6 +82,9 @@ protected Settings nodeSettings(int nodeOrdinal) {
         MockSecureSettings secureSettings = new MockSecureSettings();
         secureSettings.setString(AwsEc2Service.ACCESS_KEY_SETTING.getKey(), "some_access");
         secureSettings.setString(AwsEc2Service.SECRET_KEY_SETTING.getKey(), "some_secret");
+        if (randomBoolean()) {
+            secureSettings.setString(AwsEc2Service.SESSION_TOKEN_SETTING.getKey(), "some_token");
+        }
         return Settings.builder().put(super.nodeSettings(nodeOrdinal))
             .put(DiscoveryModule.DISCOVERY_HOSTS_PROVIDER_SETTING.getKey(), "ec2")
             .put("path.logs", resolve)

plugins/repository-s3/src/main/java/org/elasticsearch/repositories/s3/InternalAwsS3Service.java

@@ -111,9 +111,7 @@ static ClientConfiguration buildConfiguration(S3ClientSettings clientSettings) {
     // pkg private for tests
     static AWSCredentialsProvider buildCredentials(Logger logger, DeprecationLogger deprecationLogger,
                                                    S3ClientSettings clientSettings, Settings repositorySettings) {
-
-
-        BasicAWSCredentials credentials = clientSettings.credentials;
+        AWSCredentials credentials = clientSettings.credentials;
         if (S3Repository.ACCESS_KEY_SETTING.exists(repositorySettings)) {
             if (S3Repository.SECRET_KEY_SETTING.exists(repositorySettings) == false) {
                 throw new IllegalArgumentException("Repository setting [" + S3Repository.ACCESS_KEY_SETTING.getKey() +

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

@@ -27,7 +27,9 @@
 
 import com.amazonaws.ClientConfiguration;
 import com.amazonaws.Protocol;
+import com.amazonaws.auth.AWSCredentials;
 import com.amazonaws.auth.BasicAWSCredentials;
+import com.amazonaws.auth.BasicSessionCredentials;
 import org.elasticsearch.common.settings.SecureSetting;
 import org.elasticsearch.common.settings.SecureString;
 import org.elasticsearch.common.settings.Setting;
@@ -51,6 +53,10 @@ class S3ClientSettings {
     static final Setting.AffixSetting<SecureString> SECRET_KEY_SETTING = Setting.affixKeySetting(PREFIX, "secret_key",
         key -> SecureSetting.secureString(key, null));
 
+    /** The (optional) session token for connecting to s3. */
+    static final Setting.AffixSetting<SecureString> SESSION_TOKEN_SETTING = Setting.affixKeySetting(PREFIX, "session_token",
+        key -> SecureSetting.secureString(key, null));
+
     /** An override for the s3 endpoint to connect to. */
     static final Setting.AffixSetting<String> ENDPOINT_SETTING = Setting.affixKeySetting(PREFIX, "endpoint",
         key -> new Setting<>(key, "", s -> s.toLowerCase(Locale.ROOT), Property.NodeScope));
@@ -88,7 +94,7 @@ class S3ClientSettings {
         key -> Setting.boolSetting(key, ClientConfiguration.DEFAULT_THROTTLE_RETRIES, Property.NodeScope));
 
     /** Credentials to authenticate with s3. */
-    final BasicAWSCredentials credentials;
+    final AWSCredentials credentials;
 
     /** The s3 endpoint the client should talk to, or empty string to use the default. */
     final String endpoint;
@@ -119,7 +125,7 @@ class S3ClientSettings {
     /** Whether the s3 client should use an exponential backoff retry policy. */
     final boolean throttleRetries;
 
-    private S3ClientSettings(BasicAWSCredentials credentials, String endpoint, Protocol protocol,
+    private S3ClientSettings(AWSCredentials credentials, String endpoint, Protocol protocol,
                              String proxyHost, int proxyPort, String proxyUsername, String proxyPassword,
                              int readTimeoutMillis, int maxRetries, boolean throttleRetries) {
         this.credentials = credentials;
@@ -158,17 +164,28 @@ static Map<String, S3ClientSettings> load(Settings settings) {
     static S3ClientSettings getClientSettings(Settings settings, String clientName) {
         try (SecureString accessKey = getConfigValue(settings, clientName, ACCESS_KEY_SETTING);
              SecureString secretKey = getConfigValue(settings, clientName, SECRET_KEY_SETTING);
+             SecureString sessionToken = getConfigValue(settings, clientName, SESSION_TOKEN_SETTING);
              SecureString proxyUsername = getConfigValue(settings, clientName, PROXY_USERNAME_SETTING);
              SecureString proxyPassword = getConfigValue(settings, clientName, PROXY_PASSWORD_SETTING)) {
-            BasicAWSCredentials credentials = null;
+            final AWSCredentials credentials;
             if (accessKey.length() != 0) {
                 if (secretKey.length() != 0) {
-                    credentials = new BasicAWSCredentials(accessKey.toString(), secretKey.toString());
+                    if (sessionToken.length() != 0) {
+                        credentials = new BasicSessionCredentials(accessKey.toString(), secretKey.toString(), sessionToken.toString());
+                    } else {
+                        credentials = new BasicAWSCredentials(accessKey.toString(), secretKey.toString());
+                    }
                 } else {
                     throw new IllegalArgumentException("Missing secret key for s3 client [" + clientName + "]");
                 }
-            } else if (secretKey.length() != 0) {
-                throw new IllegalArgumentException("Missing access key for s3 client [" + clientName + "]");
+            } else {
+                if (secretKey.length() != 0) {
+                    throw new IllegalArgumentException("Missing access key for s3 client [" + clientName + "]");
+                }
+                if (sessionToken.length() != 0) {
+                    throw new IllegalArgumentException("Missing access key and secret key for s3 client [" + clientName + "]");
+                }
+                credentials = null;
             }
             return new S3ClientSettings(
                 credentials,

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

@@ -82,6 +82,7 @@ public List<Setting<?>> getSettings() {
             // named s3 client configuration settings
             S3ClientSettings.ACCESS_KEY_SETTING,
             S3ClientSettings.SECRET_KEY_SETTING,
+            S3ClientSettings.SESSION_TOKEN_SETTING,
             S3ClientSettings.ENDPOINT_SETTING,
             S3ClientSettings.PROTOCOL_SETTING,
             S3ClientSettings.PROXY_HOST_SETTING,