Use more precise does S3 bucket exist method (#34123)

We are using a deprecated method for checking if an S3 bucket
exists. This deprecated method has a limitation that it can not
distinguish between invalid credentials and a lack of permissions. This
commit switches to using a method that correctly surfaces if invalid
credentials are supplied when checking for the existence of a bucket.

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

@@ -19,9 +19,11 @@
 
 package org.elasticsearch.repositories.s3;
 
+import com.amazonaws.AmazonServiceException;
 import com.amazonaws.services.s3.model.CannedAccessControlList;
 import com.amazonaws.services.s3.model.DeleteObjectsRequest;
 import com.amazonaws.services.s3.model.DeleteObjectsRequest.KeyVersion;
+import com.amazonaws.services.s3.model.HeadBucketRequest;
 import com.amazonaws.services.s3.model.ObjectListing;
 import com.amazonaws.services.s3.model.S3ObjectSummary;
 import com.amazonaws.services.s3.model.StorageClass;
@@ -66,14 +68,23 @@ class S3BlobStore extends AbstractComponent implements BlobStore {
 
         // Note: the method client.doesBucketExist() may return 'true' is the bucket exists
         // but we don't have access to it (ie, 403 Forbidden response code)
-        // Also, if invalid security credentials are used to execute this method, the
-        // client is not able to distinguish between bucket permission errors and
-        // invalid credential errors, and this method could return an incorrect result.
         try (AmazonS3Reference clientReference = clientReference()) {
             SocketAccess.doPrivilegedVoid(() -> {
-                if (clientReference.client().doesBucketExist(bucket) == false) {
-                    throw new IllegalArgumentException("The bucket [" + bucket + "] does not exist. Please create it before "
-                            + " creating an s3 snapshot repository backed by it.");
+                try {
+                    clientReference.client().headBucket(new HeadBucketRequest(bucket));
+                } catch (final AmazonServiceException e) {
+                    if (e.getStatusCode() == 301) {
+                        throw new IllegalArgumentException("the bucket [" + bucket + "] is in a different region than you configured", e);
+                    } else if (e.getStatusCode() == 403) {
+                        throw new IllegalArgumentException("you do not have permissions to access the bucket [" + bucket + "]", e);
+                    } else if (e.getStatusCode() == 404) {
+                        throw new IllegalArgumentException(
+                                "the bucket [" + bucket + "] does not exist;"
+                                        + " please create it before creating an S3 snapshot repository backed by it",
+                                e);
+                    } else {
+                        throw new IllegalArgumentException("error checking the existence of bucket [" + bucket + "]", e);
+                    }
                 }
             });
         }
@@ -158,7 +169,9 @@ public CannedAccessControlList getCannedACL() {
         return cannedACL;
     }
 
-    public StorageClass getStorageClass() { return storageClass; }
+    public StorageClass getStorageClass() {
+        return storageClass;
+    }
 
     public static StorageClass initStorageClass(String storageClass) {
         if ((storageClass == null) || storageClass.equals("")) {

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

@@ -20,13 +20,16 @@
 package org.elasticsearch.repositories.s3;
 
 import com.amazonaws.AmazonClientException;
+import com.amazonaws.AmazonServiceException;
 import com.amazonaws.SdkClientException;
 import com.amazonaws.services.s3.AbstractAmazonS3;
 import com.amazonaws.services.s3.model.AmazonS3Exception;
 import com.amazonaws.services.s3.model.DeleteObjectRequest;
 import com.amazonaws.services.s3.model.DeleteObjectsRequest;
 import com.amazonaws.services.s3.model.DeleteObjectsResult;
 import com.amazonaws.services.s3.model.GetObjectRequest;
+import com.amazonaws.services.s3.model.HeadBucketRequest;
+import com.amazonaws.services.s3.model.HeadBucketResult;
 import com.amazonaws.services.s3.model.ListObjectsRequest;
 import com.amazonaws.services.s3.model.ObjectListing;
 import com.amazonaws.services.s3.model.ObjectMetadata;
@@ -73,8 +76,15 @@ class MockAmazonS3 extends AbstractAmazonS3 {
     }
 
     @Override
-    public boolean doesBucketExist(final String bucket) {
-        return this.bucket.equalsIgnoreCase(bucket);
+    public HeadBucketResult headBucket(final HeadBucketRequest headBucketRequest) throws SdkClientException, AmazonServiceException {
+        if (this.bucket.equalsIgnoreCase(headBucketRequest.getBucketName())) {
+            return new HeadBucketResult();
+        } else {
+            final AmazonServiceException e =
+                    new AmazonServiceException("bucket [" + headBucketRequest.getBucketName() + "] does not exist");
+            e.setStatusCode(404);
+            throw e;
+        }
     }
 
     @Override

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

@@ -19,9 +19,13 @@
 
 package org.elasticsearch.repositories.s3;
 
+import com.amazonaws.AmazonClientException;
+import com.amazonaws.AmazonServiceException;
 import com.amazonaws.auth.AWSCredentials;
 import com.amazonaws.auth.AWSCredentialsProvider;
 import com.amazonaws.services.s3.AmazonS3;
+import com.amazonaws.services.s3.model.HeadBucketRequest;
+import com.amazonaws.services.s3.model.HeadBucketResult;
 import org.elasticsearch.cluster.metadata.RepositoryMetaData;
 import org.elasticsearch.common.SuppressForbidden;
 import org.elasticsearch.common.settings.MockSecureSettings;
@@ -57,9 +61,10 @@ static final class ClientAndCredentials extends AmazonS3Wrapper {
             }
 
             @Override
-            public boolean doesBucketExist(String bucketName) {
-                return true;
+            public HeadBucketResult headBucket(HeadBucketRequest headBucketRequest) throws AmazonClientException, AmazonServiceException {
+                return new HeadBucketResult();
             }
+
         }
 
         static final class ProxyS3Service extends S3Service {

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

@@ -19,7 +19,11 @@
 
 package org.elasticsearch.repositories.s3;
 
+import com.amazonaws.AmazonServiceException;
+import com.amazonaws.SdkClientException;
 import com.amazonaws.services.s3.AbstractAmazonS3;
+import com.amazonaws.services.s3.model.HeadBucketRequest;
+import com.amazonaws.services.s3.model.HeadBucketResult;
 import org.elasticsearch.cluster.metadata.RepositoryMetaData;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.ByteSizeUnit;
@@ -42,8 +46,8 @@ public class S3RepositoryTests extends ESTestCase {
     private static class DummyS3Client extends AbstractAmazonS3 {
 
         @Override
-        public boolean doesBucketExist(String bucketName) {
-            return true;
+        public HeadBucketResult headBucket(final HeadBucketRequest request) throws SdkClientException, AmazonServiceException {
+            return new HeadBucketResult();
         }
 
         @Override