Add Region and Signer Algorithm Overrides to S3 Repos (#52112) (#52562)

Exposes S3 SDK signing region and algorithm override settings as requested in #51861.

Closes #51861

Author: original-brownbear

docs/plugins/repository-s3.asciidoc

@@ -184,6 +184,22 @@ pattern then you should set this setting to `true` when upgrading.
     https://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/services/s3/AmazonS3Builder.html#disableChunkedEncoding--[AWS
     Java SDK documentation] for details. Defaults to `false`.
 
+`region`::
+
+    Allows specifying the signing region to use. Specificing this setting manually should not be necessary for most use cases. Generally,
+    the SDK will correctly guess the signing region to use. It should be considered an expert level setting to support S3-compatible APIs
+    that require https://docs.aws.amazon.com/general/latest/gr/signature-version-4.html[v4 signatures] and use a region other than the
+    default `us-east-1`. Defaults to empty string which means that the SDK will try to automatically determine the correct signing region.
+
+`signer_override`::
+
+    Allows specifying the name of the signature algorithm to use for signing requests by the S3 client. Specifying this setting should not
+    be necessary for most use cases. It should be considered an expert level setting to support S3-compatible APIs that do not support the
+    signing algorithm that the SDK automatically determines for them.
+    See the
+    https://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/ClientConfiguration.html#setSignerOverride-java.lang.String-[AWS
+    Java SDK documentation] for details. Defaults to empty string which means that no signing algorithm override will be used.
+
 [float]
 [[repository-s3-compatible-services]]
 ===== S3-compatible services

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

@@ -35,6 +35,7 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
+import java.util.function.Function;
 
 /**
  * A container for settings used to create an S3 client.
@@ -103,6 +104,14 @@ final class S3ClientSettings {
     static final Setting.AffixSetting<Boolean> DISABLE_CHUNKED_ENCODING = Setting.affixKeySetting(PREFIX, "disable_chunked_encoding",
         key -> Setting.boolSetting(key, false, Property.NodeScope));
 
+    /** An override for the s3 region to use for signing requests. */
+    static final Setting.AffixSetting<String> REGION = Setting.affixKeySetting(PREFIX, "region",
+        key -> new Setting<>(key, "", Function.identity(), Property.NodeScope));
+
+    /** An override for the signer to use. */
+    static final Setting.AffixSetting<String> SIGNER_OVERRIDE = Setting.affixKeySetting(PREFIX, "signer_override",
+        key -> new Setting<>(key, "", Function.identity(), Property.NodeScope));
+
     /** Credentials to authenticate with s3. */
     final S3BasicCredentials credentials;
 
@@ -141,10 +150,16 @@ final class S3ClientSettings {
     /** Whether chunked encoding should be disabled or not. */
     final boolean disableChunkedEncoding;
 
+    /** Region to use for signing requests or empty string to use default. */
+    final String region;
+
+    /** Signer override to use or empty string to use default. */
+    final String signerOverride;
+
     private S3ClientSettings(S3BasicCredentials credentials, String endpoint, Protocol protocol,
                              String proxyHost, int proxyPort, String proxyUsername, String proxyPassword,
                              int readTimeoutMillis, int maxRetries, boolean throttleRetries,
-                             boolean pathStyleAccess, boolean disableChunkedEncoding) {
+                             boolean pathStyleAccess, boolean disableChunkedEncoding, String region, String signerOverride) {
         this.credentials = credentials;
         this.endpoint = endpoint;
         this.protocol = protocol;
@@ -157,6 +172,8 @@ private S3ClientSettings(S3BasicCredentials credentials, String endpoint, Protoc
         this.throttleRetries = throttleRetries;
         this.pathStyleAccess = pathStyleAccess;
         this.disableChunkedEncoding = disableChunkedEncoding;
+        this.region = region;
+        this.signerOverride = signerOverride;
     }
 
     /**
@@ -188,10 +205,13 @@ S3ClientSettings refine(RepositoryMetaData metadata) {
         } else {
             newCredentials = credentials;
         }
+        final String newRegion = getRepoSettingOrDefault(REGION, normalizedSettings, region);
+        final String newSignerOverride = getRepoSettingOrDefault(SIGNER_OVERRIDE, normalizedSettings, signerOverride);
         if (Objects.equals(endpoint, newEndpoint) && protocol == newProtocol && Objects.equals(proxyHost, newProxyHost)
             && proxyPort == newProxyPort && newReadTimeoutMillis == readTimeoutMillis && maxRetries == newMaxRetries
             && newThrottleRetries == throttleRetries && Objects.equals(credentials, newCredentials)
-            && newDisableChunkedEncoding == disableChunkedEncoding) {
+            && newDisableChunkedEncoding == disableChunkedEncoding
+            && Objects.equals(region, newRegion) && Objects.equals(signerOverride, newSignerOverride)) {
             return this;
         }
         return new S3ClientSettings(
@@ -206,7 +226,9 @@ S3ClientSettings refine(RepositoryMetaData metadata) {
             newMaxRetries,
             newThrottleRetries,
             usePathStyleAccess,
-            newDisableChunkedEncoding
+            newDisableChunkedEncoding,
+            newRegion,
+            newSignerOverride
         );
     }
 
@@ -295,7 +317,9 @@ static S3ClientSettings getClientSettings(final Settings settings, final String
                 getConfigValue(settings, clientName, MAX_RETRIES_SETTING),
                 getConfigValue(settings, clientName, USE_THROTTLE_RETRIES_SETTING),
                 getConfigValue(settings, clientName, USE_PATH_STYLE_ACCESS),
-                getConfigValue(settings, clientName, DISABLE_CHUNKED_ENCODING)
+                getConfigValue(settings, clientName, DISABLE_CHUNKED_ENCODING),
+                getConfigValue(settings, clientName, REGION),
+                getConfigValue(settings, clientName, SIGNER_OVERRIDE)
             );
         }
     }
@@ -319,13 +343,15 @@ public boolean equals(final Object o) {
             Objects.equals(proxyHost, that.proxyHost) &&
             Objects.equals(proxyUsername, that.proxyUsername) &&
             Objects.equals(proxyPassword, that.proxyPassword) &&
-            Objects.equals(disableChunkedEncoding, that.disableChunkedEncoding);
+            Objects.equals(disableChunkedEncoding, that.disableChunkedEncoding) &&
+            Objects.equals(region, that.region) &&
+            Objects.equals(signerOverride, that.signerOverride);
     }
 
     @Override
     public int hashCode() {
         return Objects.hash(credentials, endpoint, protocol, proxyHost, proxyPort, proxyUsername, proxyPassword,
-            readTimeoutMillis, maxRetries, throttleRetries, disableChunkedEncoding);
+            readTimeoutMillis, maxRetries, throttleRetries, disableChunkedEncoding, region, signerOverride);
     }
 
     private static <T> T getConfigValue(Settings settings, String clientName,

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

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

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

@@ -141,7 +141,8 @@ AmazonS3 buildClient(final S3ClientSettings clientSettings) {
         builder.withClientConfiguration(buildConfiguration(clientSettings));
 
         final String endpoint = Strings.hasLength(clientSettings.endpoint) ? clientSettings.endpoint : Constants.S3_HOSTNAME;
-        logger.debug("using endpoint [{}]", endpoint);
+        final String region = Strings.hasLength(clientSettings.region) ? clientSettings.region : null;
+        logger.debug("using endpoint [{}] and region [{}]", endpoint, region);
 
         // If the endpoint configuration isn't set on the builder then the default behaviour is to try
         // and work out what region we are in and use an appropriate endpoint - see AwsClientBuilder#setRegion.
@@ -151,7 +152,7 @@ AmazonS3 buildClient(final S3ClientSettings clientSettings) {
         //
         // We do this because directly constructing the client is deprecated (was already deprecated in 1.1.223 too)
         // so this change removes that usage of a deprecated API.
-        builder.withEndpointConfiguration(new AwsClientBuilder.EndpointConfiguration(endpoint, null));
+        builder.withEndpointConfiguration(new AwsClientBuilder.EndpointConfiguration(endpoint, region));
         if (clientSettings.pathStyleAccess) {
             builder.enablePathStyleAccess();
         }
@@ -177,6 +178,10 @@ static ClientConfiguration buildConfiguration(S3ClientSettings clientSettings) {
             clientConfiguration.setProxyPassword(clientSettings.proxyPassword);
         }
 
+        if (Strings.hasLength(clientSettings.signerOverride)) {
+            clientConfiguration.setSignerOverride(clientSettings.signerOverride);
+        }
+
         clientConfiguration.setMaxErrorRetry(clientSettings.maxRetries);
         clientConfiguration.setUseThrottleRetries(clientSettings.throttleRetries);
         clientConfiguration.setSocketTimeout(clientSettings.readTimeoutMillis);
@@ -231,5 +236,4 @@ public void refresh() {
     public void close() {
         releaseCachedClients();
     }
-
 }

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

@@ -46,6 +46,7 @@
 import org.elasticsearch.snapshots.SnapshotId;
 import org.elasticsearch.snapshots.SnapshotsService;
 import org.elasticsearch.snapshots.mockstore.BlobStoreWrapper;
+import org.elasticsearch.test.ESIntegTestCase;
 import org.elasticsearch.threadpool.ThreadPool;
 
 import java.io.IOException;
@@ -56,14 +57,34 @@
 import java.util.List;
 import java.util.Map;
 
+import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.greaterThan;
 import static org.hamcrest.Matchers.lessThan;
+import static org.hamcrest.Matchers.startsWith;
 
 @SuppressForbidden(reason = "this test uses a HttpServer to emulate an S3 endpoint")
+// Need to set up a new cluster for each test because cluster settings use randomized authentication settings
+@ESIntegTestCase.ClusterScope(scope = ESIntegTestCase.Scope.TEST)
 public class S3BlobStoreRepositoryTests extends ESMockAPIBasedRepositoryIntegTestCase {
 
     private static final TimeValue TEST_COOLDOWN_PERIOD = TimeValue.timeValueSeconds(5L);
 
+    private String region;
+    private String signerOverride;
+
+    @Override
+    public void setUp() throws Exception {
+        if (randomBoolean()) {
+            region = "test-region";
+        }
+        if (region != null && randomBoolean()) {
+            signerOverride = randomFrom("AWS3SignerType", "AWS4SignerType");
+        } else if (randomBoolean()) {
+            signerOverride = "AWS3SignerType";
+        }
+        super.setUp();
+    }
+
     @Override
     protected String repositoryType() {
         return S3Repository.TYPE;
@@ -101,16 +122,23 @@ protected Settings nodeSettings(int nodeOrdinal) {
         secureSettings.setString(S3ClientSettings.ACCESS_KEY_SETTING.getConcreteSettingForNamespace("test").getKey(), "access");
         secureSettings.setString(S3ClientSettings.SECRET_KEY_SETTING.getConcreteSettingForNamespace("test").getKey(), "secret");
 
-        return Settings.builder()
+        final Settings.Builder builder = Settings.builder()
             .put(ThreadPool.ESTIMATED_TIME_INTERVAL_SETTING.getKey(), 0) // We have tests that verify an exact wait time
             .put(S3ClientSettings.ENDPOINT_SETTING.getConcreteSettingForNamespace("test").getKey(), httpServerUrl())
             // Disable chunked encoding as it simplifies a lot the request parsing on the httpServer side
             .put(S3ClientSettings.DISABLE_CHUNKED_ENCODING.getConcreteSettingForNamespace("test").getKey(), true)
             // Disable request throttling because some random values in tests might generate too many failures for the S3 client
             .put(S3ClientSettings.USE_THROTTLE_RETRIES_SETTING.getConcreteSettingForNamespace("test").getKey(), false)
             .put(super.nodeSettings(nodeOrdinal))
-            .setSecureSettings(secureSettings)
-            .build();
+            .setSecureSettings(secureSettings);
+
+        if (signerOverride != null) {
+            builder.put(S3ClientSettings.SIGNER_OVERRIDE.getConcreteSettingForNamespace("test").getKey(), signerOverride);
+        }
+        if (region != null) {
+            builder.put(S3ClientSettings.REGION.getConcreteSettingForNamespace("test").getKey(), region);
+        }
+        return builder.build();
     }
 
     public void testEnforcedCooldownPeriod() throws IOException {
@@ -192,11 +220,31 @@ void ensureMultiPartUploadSize(long blobSize) {
     }
 
     @SuppressForbidden(reason = "this test uses a HttpHandler to emulate an S3 endpoint")
-    private static class S3BlobStoreHttpHandler extends S3HttpHandler implements BlobStoreHttpHandler {
+    private class S3BlobStoreHttpHandler extends S3HttpHandler implements BlobStoreHttpHandler {
 
         S3BlobStoreHttpHandler(final String bucket) {
             super(bucket);
         }
+
+        @Override
+        public void handle(final HttpExchange exchange) throws IOException {
+            validateAuthHeader(exchange);
+            super.handle(exchange);
+        }
+
+        private void validateAuthHeader(HttpExchange exchange) {
+            final String authorizationHeaderV4 = exchange.getRequestHeaders().getFirst("Authorization");
+            final String authorizationHeaderV3 = exchange.getRequestHeaders().getFirst("X-amzn-authorization");
+
+            if ("AWS3SignerType".equals(signerOverride)) {
+                assertThat(authorizationHeaderV3, startsWith("AWS3"));
+            } else if ("AWS4SignerType".equals(signerOverride)) {
+                assertThat(authorizationHeaderV4, containsString("aws4_request"));
+            }
+            if (region != null && authorizationHeaderV4 != null) {
+                assertThat(authorizationHeaderV4, containsString("/" + region + "/s3/"));
+            }
+        }
     }
 
     /**

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

@@ -21,6 +21,7 @@
 
 import com.amazonaws.ClientConfiguration;
 import com.amazonaws.Protocol;
+import com.amazonaws.services.s3.AmazonS3Client;
 import org.elasticsearch.cluster.metadata.RepositoryMetaData;
 import org.elasticsearch.common.settings.MockSecureSettings;
 import org.elasticsearch.common.settings.Settings;
@@ -158,4 +159,28 @@ public void testUseChunkedEncodingCanBeSet() {
         assertThat(settings.get("default").disableChunkedEncoding, is(false));
         assertThat(settings.get("other").disableChunkedEncoding, is(true));
     }
+
+    public void testRegionCanBeSet() {
+        final String region = randomAlphaOfLength(5);
+        final Map<String, S3ClientSettings> settings = S3ClientSettings.load(
+            Settings.builder().put("s3.client.other.region", region).build());
+        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"));
+            assertThat(other.getSignerRegionOverride(), is(region));
+        }
+    }
+
+    public void testSignerOverrideCanBeSet() {
+        final String signerOverride = randomAlphaOfLength(5);
+        final Map<String, S3ClientSettings> settings = S3ClientSettings.load(
+            Settings.builder().put("s3.client.other.signer_override", signerOverride).build());
+        assertThat(settings.get("default").region, is(""));
+        assertThat(settings.get("other").signerOverride, is(signerOverride));
+        ClientConfiguration defaultConfiguration = S3Service.buildConfiguration(settings.get("default"));
+        assertThat(defaultConfiguration.getSignerOverride(), nullValue());
+        ClientConfiguration configuration = S3Service.buildConfiguration(settings.get("other"));
+        assertThat(configuration.getSignerOverride(), is(signerOverride));
+    }
 }