S3 Repository: Add back repository level credentials (#24609)

Specifying s3 access and secret keys inside repository settings are not
secure. However, until there is a way to dynamically update secure
settings, this is the only way to dynamically add repositories with
credentials that are not known at node startup time. This commit adds
back `access_key` and `secret_key` s3 repository settings, but protects
it with a required system property `allow_insecure_settings`.

core/src/main/java/org/elasticsearch/common/settings/SecureSetting.java

@@ -24,25 +24,25 @@ import java.security.GeneralSecurityException;
 import java.util.EnumSet;
 import java.util.Set;
 
+import org.elasticsearch.common.Booleans;
 import org.elasticsearch.common.util.ArrayUtils;
 
-
 /**
  * A secure setting.
  *
  * This class allows access to settings from the Elasticsearch keystore.
  */
 public abstract class SecureSetting<T> extends Setting<T> {
+
+    /** Determines whether legacy settings with sensitive values should be allowed. */
+    private static final boolean ALLOW_INSECURE_SETTINGS = Booleans.parseBoolean(System.getProperty("es.allow_insecure_settings", "false"));
+
     private static final Set<Property> ALLOWED_PROPERTIES = EnumSet.of(Property.Deprecated, Property.Shared);
 
     private static final Property[] FIXED_PROPERTIES = {
         Property.NodeScope
     };
 
-    private static final Property[] LEGACY_PROPERTIES = {
-        Property.NodeScope, Property.Deprecated, Property.Filtered
-    };
-
     private SecureSetting(String key, Property... properties) {
         super(key, (String)null, null, ArrayUtils.concat(properties, FIXED_PROPERTIES, Property.class));
         assert assertAllowedProperties(properties);
@@ -133,6 +133,23 @@ public abstract class SecureSetting<T> extends Setting<T> {
         };
     }
 
+    /**
+     * A setting which contains a sensitive string, but which for legacy reasons must be found outside secure settings.
+     * @see #secureString(String, Setting, Property...)
+     */
+    public static Setting<SecureString> insecureString(String name) {
+        return new Setting<SecureString>(name, "", SecureString::new, Property.Deprecated, Property.Filtered, Property.NodeScope) {
+            @Override
+            public SecureString get(Settings settings) {
+                if (ALLOW_INSECURE_SETTINGS == false && exists(settings)) {
+                    throw new IllegalArgumentException("Setting [" + name + "] is insecure, " +
+                        "but property [allow_insecure_settings] is not set");
+                }
+                return super.get(settings);
+            }
+        };
+    }
+
     /**
      * A setting which contains a file. Reading the setting opens an input stream to the file.
      *

plugins/repository-s3/build.gradle

@@ -54,6 +54,16 @@ bundlePlugin {
   }
 }
 
+additionalTest('testRepositoryCreds'){
+  include '**/RepositorySettingsCredentialsTests.class'
+  systemProperty 'es.allow_insecure_settings', 'true'
+}
+
+test {
+  // these are tested explicitly in separate test tasks
+  exclude '**/*CredentialsTests.class'
+}
+
 integTestCluster {
   keystoreSetting 's3.client.default.access_key', 'myaccesskey'
   keystoreSetting 's3.client.default.secret_key', 'mysecretkey'

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

@@ -26,6 +26,7 @@ import java.util.function.Function;
 import com.amazonaws.ClientConfiguration;
 import com.amazonaws.auth.AWSCredentials;
 import com.amazonaws.auth.AWSCredentialsProvider;
+import com.amazonaws.auth.BasicAWSCredentials;
 import com.amazonaws.auth.InstanceProfileCredentialsProvider;
 import com.amazonaws.http.IdleConnectionReaper;
 import com.amazonaws.internal.StaticCredentialsProvider;
@@ -35,6 +36,8 @@ import org.apache.logging.log4j.Logger;
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.component.AbstractLifecycleComponent;
+import org.elasticsearch.common.logging.DeprecationLogger;
+import org.elasticsearch.common.settings.SecureString;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
 
@@ -69,7 +72,7 @@ class InternalAwsS3Service extends AbstractLifecycleComponent implements AwsS3Se
 
         logger.debug("creating S3 client with client_name [{}], endpoint [{}]", clientName, clientSettings.endpoint);
 
-        AWSCredentialsProvider credentials = buildCredentials(logger, clientSettings);
+        AWSCredentialsProvider credentials = buildCredentials(logger, deprecationLogger, clientSettings, repositorySettings);
         ClientConfiguration configuration = buildConfiguration(clientSettings, repositorySettings);
 
         client = new AmazonS3Client(credentials, configuration);
@@ -106,14 +109,42 @@ class InternalAwsS3Service extends AbstractLifecycleComponent implements AwsS3Se
     }
 
     // pkg private for tests
-    static AWSCredentialsProvider buildCredentials(Logger logger, S3ClientSettings clientSettings) {
-        if (clientSettings.credentials == null) {
+    static AWSCredentialsProvider buildCredentials(Logger logger, DeprecationLogger deprecationLogger,
+                                                   S3ClientSettings clientSettings, Settings repositorySettings) {
+
+
+        BasicAWSCredentials 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() +
+                    " must be accompanied by setting [" + S3Repository.SECRET_KEY_SETTING.getKey() + "]");
+            }
+            try (SecureString key = S3Repository.ACCESS_KEY_SETTING.get(repositorySettings);
+                 SecureString secret = S3Repository.SECRET_KEY_SETTING.get(repositorySettings)) {
+                credentials = new BasicAWSCredentials(key.toString(), secret.toString());
+            }
+            // backcompat for reading keys out of repository settings
+            deprecationLogger.deprecated("Using s3 access/secret key from repository settings. Instead " +
+                "store these in named clients and the elasticsearch keystore for secure settings.");
+        } else if (S3Repository.SECRET_KEY_SETTING.exists(repositorySettings)) {
+            throw new IllegalArgumentException("Repository setting [" + S3Repository.SECRET_KEY_SETTING.getKey() +
+                " must be accompanied by setting [" + S3Repository.ACCESS_KEY_SETTING.getKey() + "]");
+        }
+        if (credentials == null) {
             logger.debug("Using instance profile credentials");
             return new PrivilegedInstanceProfileCredentialsProvider();
         } else {
             logger.debug("Using basic key/secret credentials");
-            return new StaticCredentialsProvider(clientSettings.credentials);
+            return new StaticCredentialsProvider(credentials);
+        }
+    }
+
+    /** Returns the value for a given setting from the repository, or returns the fallback value. */
+    private static <T> T getRepoValue(Settings repositorySettings, Setting<T> repositorySetting, T fallback) {
+        if (repositorySetting.exists(repositorySettings)) {
+            return repositorySetting.get(repositorySettings);
         }
+        return fallback;
     }
 
     @Override

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

@@ -21,14 +21,14 @@ package org.elasticsearch.repositories.s3;
 
 import java.io.IOException;
 
-import com.amazonaws.ClientConfiguration;
 import com.amazonaws.services.s3.AmazonS3;
 import org.elasticsearch.cluster.metadata.RepositoryMetaData;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.blobstore.BlobPath;
 import org.elasticsearch.common.blobstore.BlobStore;
+import org.elasticsearch.common.settings.SecureSetting;
+import org.elasticsearch.common.settings.SecureString;
 import org.elasticsearch.common.settings.Setting;
-import org.elasticsearch.common.settings.Setting.Property;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.ByteSizeUnit;
 import org.elasticsearch.common.unit.ByteSizeValue;
@@ -53,6 +53,12 @@ class S3Repository extends BlobStoreRepository {
 
     static final String TYPE = "s3";
 
+    /** The access key to authenticate with s3. This setting is insecure because cluster settings are stored in cluster state */
+    static final Setting<SecureString> ACCESS_KEY_SETTING = SecureSetting.insecureString("access_key");
+
+    /** The secret key to authenticate with s3. This setting is insecure because cluster settings are stored in cluster state */
+    static final Setting<SecureString> SECRET_KEY_SETTING = SecureSetting.insecureString("secret_key");
+
     /**
      * Default is to use 100MB (S3 defaults) for heaps above 2GB and 5% of
      * the available memory for smaller heaps.

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

@@ -24,10 +24,12 @@ import com.amazonaws.Protocol;
 import com.amazonaws.auth.AWSCredentials;
 import com.amazonaws.auth.AWSCredentialsProvider;
 import org.elasticsearch.common.settings.MockSecureSettings;
+import org.elasticsearch.common.settings.SecureSetting;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.test.ESTestCase;
 
+import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.is;
 
@@ -35,7 +37,8 @@ public class AwsS3ServiceImplTests extends ESTestCase {
 
     public void testAWSCredentialsWithSystemProviders() {
         S3ClientSettings clientSettings = S3ClientSettings.getClientSettings(Settings.EMPTY, "default");
-        AWSCredentialsProvider credentialsProvider = InternalAwsS3Service.buildCredentials(logger, clientSettings);
+        AWSCredentialsProvider credentialsProvider =
+            InternalAwsS3Service.buildCredentials(logger, deprecationLogger, clientSettings, Settings.EMPTY);
         assertThat(credentialsProvider, instanceOf(InternalAwsS3Service.PrivilegedInstanceProfileCredentialsProvider.class));
     }
 
@@ -44,7 +47,7 @@ public class AwsS3ServiceImplTests extends ESTestCase {
         secureSettings.setString("s3.client.default.access_key", "aws_key");
         secureSettings.setString("s3.client.default.secret_key", "aws_secret");
         Settings settings = Settings.builder().setSecureSettings(secureSettings).build();
-        launchAWSCredentialsWithElasticsearchSettingsTest(Settings.EMPTY, settings, "aws_key", "aws_secret");
+        assertCredentials(Settings.EMPTY, settings, "aws_key", "aws_secret");
     }
 
     public void testAwsCredsExplicitConfigSettings() {
@@ -55,14 +58,38 @@ public class AwsS3ServiceImplTests extends ESTestCase {
         secureSettings.setString("s3.client.default.access_key", "wrong_key");
         secureSettings.setString("s3.client.default.secret_key", "wrong_secret");
         Settings settings = Settings.builder().setSecureSettings(secureSettings).build();
-        launchAWSCredentialsWithElasticsearchSettingsTest(repositorySettings, settings, "aws_key", "aws_secret");
+        assertCredentials(repositorySettings, settings, "aws_key", "aws_secret");
     }
 
-    private void launchAWSCredentialsWithElasticsearchSettingsTest(Settings singleRepositorySettings, Settings settings,
-                                                                     String expectedKey, String expectedSecret) {
+    public void testRepositorySettingsCredentialsDisallowed() {
+        Settings repositorySettings = Settings.builder()
+            .put(S3Repository.ACCESS_KEY_SETTING.getKey(), "aws_key")
+            .put(S3Repository.SECRET_KEY_SETTING.getKey(), "aws_secret").build();
+        IllegalArgumentException e = expectThrows(IllegalArgumentException.class, () ->
+            assertCredentials(repositorySettings, Settings.EMPTY, "aws_key", "aws_secret"));
+        assertThat(e.getMessage(), containsString("Setting [access_key] is insecure"));
+    }
+
+    public void testRepositorySettingsCredentialsMissingKey() {
+        Settings repositorySettings = Settings.builder().put(S3Repository.SECRET_KEY_SETTING.getKey(), "aws_secret").build();
+        IllegalArgumentException e = expectThrows(IllegalArgumentException.class, () ->
+            assertCredentials(repositorySettings, Settings.EMPTY, "aws_key", "aws_secret"));
+        assertThat(e.getMessage(), containsString("must be accompanied by setting [access_key]"));
+    }
+
+    public void testRepositorySettingsCredentialsMissingSecret() {
+        Settings repositorySettings = Settings.builder().put(S3Repository.ACCESS_KEY_SETTING.getKey(), "aws_key").build();
+        IllegalArgumentException e = expectThrows(IllegalArgumentException.class, () ->
+            assertCredentials(repositorySettings, Settings.EMPTY, "aws_key", "aws_secret"));
+        assertThat(e.getMessage(), containsString("must be accompanied by setting [secret_key]"));
+    }
+
+    private void assertCredentials(Settings singleRepositorySettings, Settings settings,
+                                   String expectedKey, String expectedSecret) {
         String configName = InternalAwsS3Service.CLIENT_NAME.get(singleRepositorySettings);
         S3ClientSettings clientSettings = S3ClientSettings.getClientSettings(settings, configName);
-        AWSCredentials credentials = InternalAwsS3Service.buildCredentials(logger, clientSettings).getCredentials();
+        AWSCredentials credentials = InternalAwsS3Service.buildCredentials(logger, deprecationLogger,
+            clientSettings, singleRepositorySettings).getCredentials();
         assertThat(credentials.getAWSAccessKeyId(), is(expectedKey));
         assertThat(credentials.getAWSSecretKey(), is(expectedSecret));
     }

plugins/repository-s3/src/test/java/org/elasticsearch/repositories/s3/RepositorySettingsCredentialsTests.java

@@ -0,0 +1,41 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.repositories.s3;
+
+import com.amazonaws.auth.AWSCredentials;
+import org.elasticsearch.common.settings.Setting;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.test.ESTestCase;
+
+public class RepositorySettingsCredentialsTests extends ESTestCase {
+
+    public void testRepositorySettingsCredentials() {
+        Settings repositorySettings = Settings.builder()
+            .put(S3Repository.ACCESS_KEY_SETTING.getKey(), "aws_key")
+            .put(S3Repository.SECRET_KEY_SETTING.getKey(), "aws_secret").build();
+        AWSCredentials credentials = InternalAwsS3Service.buildCredentials(logger, deprecationLogger,
+            S3ClientSettings.getClientSettings(Settings.EMPTY, "default"), repositorySettings).getCredentials();
+        assertEquals("aws_key", credentials.getAWSAccessKeyId());
+        assertEquals("aws_secret", credentials.getAWSSecretKey());
+        assertSettingDeprecationsAndWarnings(new Setting<?>[] { S3Repository.ACCESS_KEY_SETTING, S3Repository.SECRET_KEY_SETTING },
+            "Using s3 access/secret key from repository settings. " +
+                "Instead store these in named clients and the elasticsearch keystore for secure settings.");
+    }
+}