Do not override named S3 client credentials

plugins/repository-s3/build.gradle

@@ -70,12 +70,14 @@ bundlePlugin {
 
 additionalTest('testRepositoryCreds'){
   include '**/RepositoryCredentialsTests.class'
+  include '**/S3BlobStoreRepositoryTests.class'
   systemProperty 'es.allow_insecure_settings', 'true'
 }
 
 test {
   // these are tested explicitly in separate test tasks
   exclude '**/*CredentialsTests.class'
+  exclude '**/S3BlobStoreRepositoryTests.class'
 }
 
 boolean useFixture = false

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

@@ -19,25 +19,25 @@
 
 package org.elasticsearch.repositories.s3;
 
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Locale;
-import java.util.Map;
-import java.util.Set;
 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.collect.MapBuilder;
+import org.elasticsearch.cluster.metadata.RepositoryMetaData;
 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.TimeValue;
 
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+
 /**
  * A container for settings used to create an S3 client.
  */
@@ -160,19 +160,6 @@ static Map<String, S3ClientSettings> load(Settings settings) {
         return Collections.unmodifiableMap(clients);
     }
 
-    static Map<String, S3ClientSettings> overrideCredentials(Map<String, S3ClientSettings> clientsSettings,
-                                                             BasicAWSCredentials credentials) {
-        final MapBuilder<String, S3ClientSettings> mapBuilder = new MapBuilder<>();
-        for (final Map.Entry<String, S3ClientSettings> entry : clientsSettings.entrySet()) {
-            final S3ClientSettings s3ClientSettings = new S3ClientSettings(credentials, entry.getValue().endpoint,
-                    entry.getValue().protocol, entry.getValue().proxyHost, entry.getValue().proxyPort, entry.getValue().proxyUsername,
-                    entry.getValue().proxyPassword, entry.getValue().readTimeoutMillis, entry.getValue().maxRetries,
-                    entry.getValue().throttleRetries);
-            mapBuilder.put(entry.getKey(), s3ClientSettings);
-        }
-        return mapBuilder.immutableMap();
-    }
-
     static boolean checkDeprecatedCredentials(Settings repositorySettings) {
         if (S3Repository.ACCESS_KEY_SETTING.exists(repositorySettings)) {
             if (S3Repository.SECRET_KEY_SETTING.exists(repositorySettings) == false) {
@@ -224,25 +211,37 @@ static AWSCredentials loadCredentials(Settings settings, String clientName) {
 
     // pkg private for tests
     /** Parse settings for a single client. */
-    static S3ClientSettings getClientSettings(Settings settings, String clientName) {
+    static S3ClientSettings getClientSettings(final Settings settings, final String clientName) {
         final AWSCredentials credentials = S3ClientSettings.loadCredentials(settings, clientName);
+        return getClientSettings(settings, clientName, credentials);
+    }
+
+    static S3ClientSettings getClientSettings(final Settings settings, final String clientName, final AWSCredentials credentials) {
         try (SecureString proxyUsername = getConfigValue(settings, clientName, PROXY_USERNAME_SETTING);
              SecureString proxyPassword = getConfigValue(settings, clientName, PROXY_PASSWORD_SETTING)) {
             return new S3ClientSettings(
-                credentials,
-                getConfigValue(settings, clientName, ENDPOINT_SETTING),
-                getConfigValue(settings, clientName, PROTOCOL_SETTING),
-                getConfigValue(settings, clientName, PROXY_HOST_SETTING),
-                getConfigValue(settings, clientName, PROXY_PORT_SETTING),
-                proxyUsername.toString(),
-                proxyPassword.toString(),
-                (int)getConfigValue(settings, clientName, READ_TIMEOUT_SETTING).millis(),
-                getConfigValue(settings, clientName, MAX_RETRIES_SETTING),
-                getConfigValue(settings, clientName, USE_THROTTLE_RETRIES_SETTING)
+                    credentials,
+                    getConfigValue(settings, clientName, ENDPOINT_SETTING),
+                    getConfigValue(settings, clientName, PROTOCOL_SETTING),
+                    getConfigValue(settings, clientName, PROXY_HOST_SETTING),
+                    getConfigValue(settings, clientName, PROXY_PORT_SETTING),
+                    proxyUsername.toString(),
+                    proxyPassword.toString(),
+                    Math.toIntExact(getConfigValue(settings, clientName, READ_TIMEOUT_SETTING).millis()),
+                    getConfigValue(settings, clientName, MAX_RETRIES_SETTING),
+                    getConfigValue(settings, clientName, USE_THROTTLE_RETRIES_SETTING)
             );
         }
     }
 
+    static S3ClientSettings getClientSettings(final RepositoryMetaData metadata, final AWSCredentials credentials) {
+        final Settings.Builder builder = Settings.builder();
+        for (final String key : metadata.settings().keySet()) {
+            builder.put(PREFIX + "provided" + "." + key, metadata.settings().get(key));
+        }
+        return getClientSettings(builder.build(), "provided", credentials);
+    }
+
     private static <T> T getConfigValue(Settings settings, String clientName,
                                         Setting.AffixSetting<T> clientSetting) {
         final Setting<T> concreteSetting = clientSetting.getConcreteSettingForNamespace(clientName);

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

@@ -35,7 +35,6 @@
 import org.elasticsearch.repositories.RepositoryException;
 import org.elasticsearch.repositories.blobstore.BlobStoreRepository;
 
-import java.util.Map;
 import java.util.function.Function;
 
 /**
@@ -163,6 +162,8 @@ class S3Repository extends BlobStoreRepository {
 
     private final String clientName;
 
+    private final AmazonS3Reference reference;
+
     /**
      * Constructs an s3 backed repository
      */
@@ -200,21 +201,54 @@ class S3Repository extends BlobStoreRepository {
 
         this.storageClass = STORAGE_CLASS_SETTING.get(metadata.settings());
         this.cannedACL = CANNED_ACL_SETTING.get(metadata.settings());
+
         this.clientName = CLIENT_NAME.get(metadata.settings());
 
-        logger.debug("using bucket [{}], chunk_size [{}], server_side_encryption [{}], " +
-                "buffer_size [{}], cannedACL [{}], storageClass [{}]",
-            bucket, chunkSize, serverSideEncryption, bufferSize, cannedACL, storageClass);
+        if (CLIENT_NAME.exists(metadata.settings()) && S3ClientSettings.checkDeprecatedCredentials(metadata.settings())) {
+            logger.warn(
+                    "ignoring use of named client [{}] for repository [{}] as insecure credentials were specified",
+                    clientName,
+                    metadata.name());
+        }
 
-        // (repository settings)
         if (S3ClientSettings.checkDeprecatedCredentials(metadata.settings())) {
-            overrideCredentialsFromClusterState(service);
+            // provided 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.");
+            final BasicAWSCredentials insecureCredentials = S3ClientSettings.loadDeprecatedCredentials(metadata.settings());
+            final S3ClientSettings s3ClientSettings = S3ClientSettings.getClientSettings(metadata, insecureCredentials);
+            this.reference = new AmazonS3Reference(service.buildClient(s3ClientSettings));
+        } else {
+            reference = null;
         }
+
+        logger.debug(
+                "using bucket [{}], chunk_size [{}], server_side_encryption [{}], buffer_size [{}], cannedACL [{}], storageClass [{}]",
+                bucket,
+                chunkSize,
+                serverSideEncryption,
+                bufferSize,
+                cannedACL,
+                storageClass);
     }
 
     @Override
     protected S3BlobStore createBlobStore() {
-        return new S3BlobStore(settings, service, clientName, bucket, serverSideEncryption, bufferSize, cannedACL, storageClass);
+        if (reference != null) {
+            assert S3ClientSettings.checkDeprecatedCredentials(metadata.settings()) : metadata.name();
+            return new S3BlobStore(settings, service, clientName, bucket, serverSideEncryption, bufferSize, cannedACL, storageClass) {
+                @Override
+                public AmazonS3Reference clientReference() {
+                    if (reference.tryIncRef()) {
+                        return reference;
+                    } else {
+                        throw new IllegalStateException("S3 client is closed");
+                    }
+                }
+            };
+        } else {
+            return new S3BlobStore(settings, service, clientName, bucket, serverSideEncryption, bufferSize, cannedACL, storageClass);
+        }
     }
 
     // only use for testing
@@ -244,13 +278,13 @@ protected ByteSizeValue chunkSize() {
         return chunkSize;
     }
 
-    void overrideCredentialsFromClusterState(final S3Service s3Service) {
-        deprecationLogger.deprecated("Using s3 access/secret key from repository settings. Instead "
-                + "store these in named clients and the elasticsearch keystore for secure settings.");
-        final BasicAWSCredentials insecureCredentials = S3ClientSettings.loadDeprecatedCredentials(metadata.settings());
-        // hack, but that's ok because the whole if branch should be axed
-        final Map<String, S3ClientSettings> prevSettings = s3Service.refreshAndClearCache(S3ClientSettings.load(Settings.EMPTY));
-        final Map<String, S3ClientSettings> newSettings = S3ClientSettings.overrideCredentials(prevSettings, insecureCredentials);
-        s3Service.refreshAndClearCache(newSettings);
+    @Override
+    protected void doClose() {
+        if (reference != null) {
+            assert S3ClientSettings.checkDeprecatedCredentials(metadata.settings()) : metadata.name();
+            reference.decRef();
+        }
+        super.doClose();
     }
+
 }

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

@@ -107,7 +107,6 @@ public void testRepositoryCredentialsOverrideSecureCredentials() throws IOExcept
         final Settings settings = Settings.builder().setSecureSettings(secureSettings).build();
         // repository settings for credentials override node secure settings
         final RepositoryMetaData metadata = new RepositoryMetaData("dummy-repo", "mock", Settings.builder()
-                .put(S3Repository.CLIENT_NAME.getKey(), randomFrom(clientNames))
                 .put(S3Repository.ACCESS_KEY_SETTING.getKey(), "insecure_aws_key")
                 .put(S3Repository.SECRET_KEY_SETTING.getKey(), "insecure_aws_secret").build());
         try (S3RepositoryPlugin s3Plugin = new ProxyS3RepositoryPlugin(settings);
@@ -163,11 +162,13 @@ public void testReinitSecureCredentials() throws IOException {
         secureSettings.setString("s3.client." + clientName + ".secret_key", "secure_aws_secret");
         final Settings settings = Settings.builder().setSecureSettings(secureSettings).build();
         // repository settings
-        final Settings.Builder builder = Settings.builder().put(S3Repository.CLIENT_NAME.getKey(), clientName);
+        final Settings.Builder builder = Settings.builder();
         final boolean repositorySettings = randomBoolean();
         if (repositorySettings) {
             builder.put(S3Repository.ACCESS_KEY_SETTING.getKey(), "insecure_aws_key");
             builder.put(S3Repository.SECRET_KEY_SETTING.getKey(), "insecure_aws_secret");
+        } else {
+            builder.put(S3Repository.CLIENT_NAME.getKey(), clientName);
         }
         final RepositoryMetaData metadata = new RepositoryMetaData("dummy-repo", "mock", builder.build());
         try (S3RepositoryPlugin s3Plugin = new ProxyS3RepositoryPlugin(settings);
@@ -202,8 +203,13 @@ public void testReinitSecureCredentials() throws IOException {
             try (AmazonS3Reference s3Ref = ((S3BlobStore) s3repo.blobStore()).clientReference()) {
                 final AWSCredentials newCredentials = ((ProxyS3RepositoryPlugin.ClientAndCredentials) s3Ref.client()).credentials
                         .getCredentials();
-                assertThat(newCredentials.getAWSAccessKeyId(), is("new_secret_aws_key"));
-                assertThat(newCredentials.getAWSSecretKey(), is("new_secret_aws_secret"));
+                if (repositorySettings) {
+                    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 (repositorySettings) {

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

@@ -18,9 +18,9 @@
  */
 package org.elasticsearch.repositories.s3;
 
+import com.amazonaws.services.s3.AmazonS3;
 import com.amazonaws.services.s3.model.CannedAccessControlList;
 import com.amazonaws.services.s3.model.StorageClass;
-
 import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.settings.SettingsFilter;
@@ -91,7 +91,6 @@ protected void createTestRepository(final String name, boolean verify) {
             .setVerify(verify)
             .setSettings(Settings.builder()
                 .put(S3Repository.BUCKET_SETTING.getKey(), bucket)
-                .put(S3Repository.CLIENT_NAME.getKey(), client)
                 .put(S3Repository.BUFFER_SIZE_SETTING.getKey(), bufferSize)
                 .put(S3Repository.SERVER_SIDE_ENCRYPTION_SETTING.getKey(), serverSideEncryption)
                 .put(S3Repository.CANNED_ACL_SETTING.getKey(), cannedACL)
@@ -121,14 +120,10 @@ public Map<String, Repository.Factory> getRepositories(final Environment env, fi
             return Collections.singletonMap(S3Repository.TYPE,
                     (metadata) -> new S3Repository(metadata, env.settings(), registry, new S3Service(env.settings()) {
                         @Override
-                        public synchronized AmazonS3Reference client(String clientName) {
-                            return new AmazonS3Reference(new MockAmazonS3(blobs, bucket, serverSideEncryption, cannedACL, storageClass));
-                        }
-                    }) {
-                        @Override
-                        void overrideCredentialsFromClusterState(S3Service awsService) {
+                        AmazonS3 buildClient(S3ClientSettings clientSettings) {
+                            return new MockAmazonS3(blobs, bucket, serverSideEncryption, cannedACL, storageClass);
                         }
-                    });
+                    }));
         }
     }