Reintroduce the ability to configure S3 repository credentials in cluster state

modules/repository-s3/build.gradle

@@ -75,6 +75,20 @@ esplugin.bundleSpec.from('config/repository-s3') {
     into 'config'
 }
 
+def testRepositoryCreds = tasks.register("testRepositoryCreds", Test) {
+  include '**/RepositoryCredentialsTests.class'
+  systemProperty 'es.allow_insecure_settings', 'true'
+}
+
+tasks.named('check').configure {
+  dependsOn(testRepositoryCreds)
+}
+
+tasks.named('test').configure {
+  // this is tested explicitly in separate test tasks
+  exclude '**/RepositoryCredentialsTests.class'
+}
+
 boolean useFixture = false
 
 def fixtureAddress = { fixture, name, port ->

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

@@ -263,6 +263,12 @@ S3ClientSettings refine(Settings repositorySettings) {
             normalizedSettings,
             disableChunkedEncoding
         );
+        final S3BasicCredentials newCredentials;
+        if (checkDeprecatedCredentials(repositorySettings)) {
+            newCredentials = loadDeprecatedCredentials(repositorySettings);
+        } else {
+            newCredentials = credentials;
+        }
         final String newRegion = getRepoSettingOrDefault(REGION, normalizedSettings, region);
         final String newSignerOverride = getRepoSettingOrDefault(SIGNER_OVERRIDE, normalizedSettings, signerOverride);
         if (Objects.equals(endpoint, newEndpoint)
@@ -272,14 +278,15 @@ S3ClientSettings refine(Settings repositorySettings) {
             && newReadTimeoutMillis == readTimeoutMillis
             && maxRetries == newMaxRetries
             && newThrottleRetries == throttleRetries
+            && Objects.equals(credentials, newCredentials)
             && newPathStyleAccess == pathStyleAccess
             && newDisableChunkedEncoding == disableChunkedEncoding
             && Objects.equals(region, newRegion)
             && Objects.equals(signerOverride, newSignerOverride)) {
             return this;
         }
         return new S3ClientSettings(
-            credentials,
+            newCredentials,
             newEndpoint,
             newProtocol,
             newProxyHost,
@@ -315,6 +322,41 @@ static Map<String, S3ClientSettings> load(Settings settings) {
         return Collections.unmodifiableMap(clients);
     }
 
+    static boolean checkDeprecatedCredentials(Settings repositorySettings) {
+        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()
+                        + "]"
+                );
+            }
+            return true;
+        } 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()
+                    + "]"
+            );
+        }
+        return false;
+    }
+
+    // backcompat for reading keys out of repository settings (clusterState)
+    private static S3BasicCredentials loadDeprecatedCredentials(Settings repositorySettings) {
+        assert checkDeprecatedCredentials(repositorySettings);
+        try (
+            SecureString key = S3Repository.ACCESS_KEY_SETTING.get(repositorySettings);
+            SecureString secret = S3Repository.SECRET_KEY_SETTING.get(repositorySettings)
+        ) {
+            return new S3BasicCredentials(key.toString(), secret.toString());
+        }
+    }
+
     private static S3BasicCredentials loadCredentials(Settings settings, String clientName) {
         try (
             SecureString accessKey = getConfigValue(settings, clientName, ACCESS_KEY_SETTING);

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

@@ -18,6 +18,10 @@
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.blobstore.BlobPath;
 import org.elasticsearch.common.blobstore.BlobStore;
+import org.elasticsearch.common.logging.DeprecationCategory;
+import org.elasticsearch.common.logging.DeprecationLogger;
+import org.elasticsearch.common.settings.SecureSetting;
+import org.elasticsearch.common.settings.SecureString;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.unit.ByteSizeUnit;
 import org.elasticsearch.common.unit.ByteSizeValue;
@@ -57,9 +61,16 @@
  */
 class S3Repository extends MeteredBlobStoreRepository {
     private static final Logger logger = LogManager.getLogger(S3Repository.class);
+    private static final DeprecationLogger deprecationLogger = DeprecationLogger.getLogger(logger.getName());
 
     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.
@@ -233,6 +244,16 @@ class S3Repository extends MeteredBlobStoreRepository {
         this.storageClass = STORAGE_CLASS_SETTING.get(metadata.settings());
         this.cannedACL = CANNED_ACL_SETTING.get(metadata.settings());
 
+        if (S3ClientSettings.checkDeprecatedCredentials(metadata.settings())) {
+            // provided repository settings
+            deprecationLogger.critical(
+                DeprecationCategory.SECURITY,
+                "s3_repository_secret_settings",
+                "Using s3 access/secret key from repository settings. Instead "
+                    + "store these in named clients and the elasticsearch keystore for secure settings."
+            );
+        }
+
         coolDown = COOLDOWN_PERIOD.get(metadata.settings());
 
         logger.debug(

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

@@ -141,7 +141,9 @@ public List<Setting<?>> getSettings() {
             S3ClientSettings.USE_THROTTLE_RETRIES_SETTING,
             S3ClientSettings.USE_PATH_STYLE_ACCESS,
             S3ClientSettings.SIGNER_OVERRIDE,
-            S3ClientSettings.REGION
+            S3ClientSettings.REGION,
+            S3Repository.ACCESS_KEY_SETTING,
+            S3Repository.SECRET_KEY_SETTING
         );
     }
 

modules/repository-s3/src/main/plugin-metadata/plugin-security.policy

@@ -28,5 +28,6 @@ grant {
   // s3 client opens socket connections for to access repository
   permission java.net.SocketPermission "*", "connect";
 
-
+  // only for tests : org.elasticsearch.repositories.s3.S3RepositoryPlugin
+  permission java.util.PropertyPermission "es.allow_insecure_settings", "read,write";
 };

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

@@ -14,31 +14,54 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.elasticsearch.client.internal.node.NodeClient;
 import org.elasticsearch.cluster.metadata.RepositoryMetadata;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.settings.MockSecureSettings;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.settings.SettingsFilter;
 import org.elasticsearch.common.util.BigArrays;
+import org.elasticsearch.core.SuppressForbidden;
 import org.elasticsearch.env.Environment;
 import org.elasticsearch.indices.recovery.RecoverySettings;
 import org.elasticsearch.plugins.Plugin;
 import org.elasticsearch.plugins.PluginsService;
 import org.elasticsearch.repositories.RepositoriesService;
+import org.elasticsearch.rest.AbstractRestChannel;
+import org.elasticsearch.rest.RestRequest;
+import org.elasticsearch.rest.RestResponse;
+import org.elasticsearch.rest.action.admin.cluster.RestGetRepositoriesAction;
 import org.elasticsearch.test.ESSingleNodeTestCase;
+import org.elasticsearch.test.rest.FakeRestRequest;
 import org.elasticsearch.xcontent.NamedXContentRegistry;
 
+import java.security.AccessController;
+import java.security.PrivilegedAction;
 import java.util.Collection;
 import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicReference;
 
 import static org.elasticsearch.repositories.s3.S3ClientSettings.ACCESS_KEY_SETTING;
 import static org.elasticsearch.repositories.s3.S3ClientSettings.SECRET_KEY_SETTING;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
+import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.not;
 import static org.hamcrest.Matchers.notNullValue;
 
+@SuppressForbidden(reason = "test requires to set a System property to allow insecure settings when running in IDE")
 public class RepositoryCredentialsTests extends ESSingleNodeTestCase {
 
+    static {
+        AccessController.doPrivileged((PrivilegedAction<Void>) () -> {
+            // required for client settings overwriting when running in IDE
+            System.setProperty("es.allow_insecure_settings", "true");
+            return null;
+        });
+    }
+
     @Override
     protected Collection<Class<? extends Plugin>> getPlugins() {
         return List.of(ProxyS3RepositoryPlugin.class);
@@ -60,11 +83,51 @@ protected Settings nodeSettings() {
         return Settings.builder().setSecureSettings(secureSettings).put(super.nodeSettings()).build();
     }
 
+    public void testRepositoryCredentialsOverrideSecureCredentials() {
+        final String repositoryName = "repo-creds-override";
+        final Settings.Builder repositorySettings = Settings.builder()
+            // repository settings for credentials override node secure settings
+            .put(S3Repository.ACCESS_KEY_SETTING.getKey(), "insecure_aws_key")
+            .put(S3Repository.SECRET_KEY_SETTING.getKey(), "insecure_aws_secret");
+
+        final String clientName = randomFrom("default", "other", null);
+        if (clientName != null) {
+            repositorySettings.put(S3Repository.CLIENT_NAME.getKey(), clientName);
+        }
+        createRepository(repositoryName, repositorySettings.build());
+
+        final RepositoriesService repositories = getInstanceFromNode(RepositoriesService.class);
+        assertThat(repositories.repository(repositoryName), notNullValue());
+        assertThat(repositories.repository(repositoryName), instanceOf(S3Repository.class));
+
+        final S3Repository repository = (S3Repository) repositories.repository(repositoryName);
+        final AmazonS3 client = repository.createBlobStore().clientReference().client();
+        assertThat(client, instanceOf(ProxyS3RepositoryPlugin.ClientAndCredentials.class));
+
+        final AWSCredentials credentials = ((ProxyS3RepositoryPlugin.ClientAndCredentials) client).credentials.getCredentials();
+        assertThat(credentials.getAWSAccessKeyId(), is("insecure_aws_key"));
+        assertThat(credentials.getAWSSecretKey(), is("insecure_aws_secret"));
+
+        assertCriticalWarnings(
+            "[secret_key] setting was deprecated in Elasticsearch and will be removed in a future release.",
+            "Using s3 access/secret key from repository settings. Instead store these in named clients and"
+                + " the elasticsearch keystore for secure settings.",
+            "[access_key] setting was deprecated in Elasticsearch and will be removed in a future release."
+        );
+    }
+
     public void testReinitSecureCredentials() {
         final String clientName = randomFrom("default", "other");
 
         final Settings.Builder repositorySettings = Settings.builder();
-        repositorySettings.put(S3Repository.CLIENT_NAME.getKey(), clientName);
+        final boolean hasInsecureSettings = randomBoolean();
+        if (hasInsecureSettings) {
+            // repository settings for credentials override node secure settings
+            repositorySettings.put(S3Repository.ACCESS_KEY_SETTING.getKey(), "insecure_aws_key");
+            repositorySettings.put(S3Repository.SECRET_KEY_SETTING.getKey(), "insecure_aws_secret");
+        } else {
+            repositorySettings.put(S3Repository.CLIENT_NAME.getKey(), clientName);
+        }
 
         final String repositoryName = "repo-reinit-creds";
         createRepository(repositoryName, repositorySettings.build());
@@ -79,7 +142,10 @@ public void testReinitSecureCredentials() {
             assertThat(client, instanceOf(ProxyS3RepositoryPlugin.ClientAndCredentials.class));
 
             final AWSCredentials credentials = ((ProxyS3RepositoryPlugin.ClientAndCredentials) client).credentials.getCredentials();
-            if ("other".equals(clientName)) {
+            if (hasInsecureSettings) {
+                assertThat(credentials.getAWSAccessKeyId(), is("insecure_aws_key"));
+                assertThat(credentials.getAWSSecretKey(), is("insecure_aws_secret"));
+            } else if ("other".equals(clientName)) {
                 assertThat(credentials.getAWSAccessKeyId(), is("secure_other_key"));
                 assertThat(credentials.getAWSSecretKey(), is("secure_other_secret"));
             } else {
@@ -98,7 +164,10 @@ public void testReinitSecureCredentials() {
             plugin.reload(newSettings);
 
             // check the not-yet-closed client reference still has the same credentials
-            if ("other".equals(clientName)) {
+            if (hasInsecureSettings) {
+                assertThat(credentials.getAWSAccessKeyId(), is("insecure_aws_key"));
+                assertThat(credentials.getAWSSecretKey(), is("insecure_aws_secret"));
+            } else if ("other".equals(clientName)) {
                 assertThat(credentials.getAWSAccessKeyId(), is("secure_other_key"));
                 assertThat(credentials.getAWSSecretKey(), is("secure_other_secret"));
             } else {
@@ -113,11 +182,66 @@ public void testReinitSecureCredentials() {
             assertThat(client, instanceOf(ProxyS3RepositoryPlugin.ClientAndCredentials.class));
 
             final AWSCredentials newCredentials = ((ProxyS3RepositoryPlugin.ClientAndCredentials) client).credentials.getCredentials();
-            assertThat(newCredentials.getAWSAccessKeyId(), is("new_secret_aws_key"));
-            assertThat(newCredentials.getAWSSecretKey(), is("new_secret_aws_secret"));
+            if (hasInsecureSettings) {
+                assertThat(newCredentials.getAWSAccessKeyId(), is("insecure_aws_key"));
+                assertThat(newCredentials.getAWSSecretKey(), is("insecure_aws_secret"));
+            } else {
+                assertThat(newCredentials.getAWSAccessKeyId(), is("new_secret_aws_key"));
+                assertThat(newCredentials.getAWSSecretKey(), is("new_secret_aws_secret"));
+            }
+        }
+
+        if (hasInsecureSettings) {
+            assertCriticalWarnings(
+                "[secret_key] setting was deprecated in Elasticsearch and will be removed in a future release.",
+                "Using s3 access/secret key from repository settings. Instead store these in named clients and"
+                    + " the elasticsearch keystore for secure settings.",
+                "[access_key] setting was deprecated in Elasticsearch and will be removed in a future release."
+            );
         }
     }
 
+    public void testInsecureRepositoryCredentials() throws Exception {
+        final String repositoryName = "repo-insecure-creds";
+        createRepository(
+            repositoryName,
+            Settings.builder()
+                .put(S3Repository.ACCESS_KEY_SETTING.getKey(), "insecure_aws_key")
+                .put(S3Repository.SECRET_KEY_SETTING.getKey(), "insecure_aws_secret")
+                .build()
+        );
+
+        final RestRequest fakeRestRequest = new FakeRestRequest();
+        fakeRestRequest.params().put("repository", repositoryName);
+        final RestGetRepositoriesAction action = new RestGetRepositoriesAction(getInstanceFromNode(SettingsFilter.class));
+
+        final CountDownLatch latch = new CountDownLatch(1);
+        final AtomicReference<AssertionError> error = new AtomicReference<>();
+        action.handleRequest(fakeRestRequest, new AbstractRestChannel(fakeRestRequest, true) {
+            @Override
+            public void sendResponse(RestResponse response) {
+                try {
+                    String responseAsString = response.content().utf8ToString();
+                    assertThat(responseAsString, containsString(repositoryName));
+                    assertThat(responseAsString, not(containsString("insecure_")));
+                } catch (final AssertionError ex) {
+                    error.set(ex);
+                }
+                latch.countDown();
+            }
+        }, getInstanceFromNode(NodeClient.class));
+
+        latch.await();
+        if (error.get() != null) {
+            throw error.get();
+        }
+
+        assertWarnings(
+            "Using s3 access/secret key from repository settings. Instead store these in named clients and"
+                + " the elasticsearch keystore for secure settings."
+        );
+    }
+
     private void createRepository(final String name, final Settings repositorySettings) {
         assertAcked(
             client().admin()