Add null and empty string validation to S3 bucket (elastic#107883)

Add basic validation to S3 bucket name - nullity and empty string. It is
aligned with public
[docs](https://www.elastic.co/guide/en/elasticsearch/reference/8.13/repository-s3.html#repository-s3-repository)
for "bucket" as required field. We might want to add more validations
based on S3 naming rules. This PR should not be a breaking change
because missing bucket  will eventually throw exception later in the
code with obscure error.

I've added yaml test to modules
[repository_s3/10_basic.yml](https://github.com/elastic/elasticsearch/compare/main...mhl-b:elasticsearch:s3-bucket-validation?expand=1#diff-08cf26742fe939f5575961254c4d3b4bff6915141cdd6abe4cd28a743d1b70ba),
not sure if it's a right place.

Addresses elastic#107840

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

@@ -223,8 +223,8 @@ class S3Repository extends MeteredBlobStoreRepository {
 
         // Parse and validate the user's S3 Storage Class setting
         this.bucket = BUCKET_SETTING.get(metadata.settings());
-        if (bucket == null) {
-            throw new RepositoryException(metadata.name(), "No bucket defined for s3 repository");
+        if (Strings.hasLength(bucket) == false) {
+            throw new IllegalArgumentException("Invalid S3 bucket name, cannot be null or empty");
         }
 
         this.bufferSize = BUFFER_SIZE_SETTING.get(metadata.settings());

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

@@ -82,6 +82,7 @@ protected Settings nodeSettings() {
     public void testRepositoryCredentialsOverrideSecureCredentials() {
         final String repositoryName = "repo-creds-override";
         final Settings.Builder repositorySettings = Settings.builder()
+            .put(S3Repository.BUCKET_SETTING.getKey(), "bucket")
             // 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");
@@ -115,7 +116,7 @@ public void testRepositoryCredentialsOverrideSecureCredentials() {
     public void testReinitSecureCredentials() {
         final String clientName = randomFrom("default", "other");
 
-        final Settings.Builder repositorySettings = Settings.builder();
+        final Settings.Builder repositorySettings = Settings.builder().put(S3Repository.BUCKET_SETTING.getKey(), "bucket");
         final boolean hasInsecureSettings = randomBoolean();
         if (hasInsecureSettings) {
             // repository settings for credentials override node secure settings
@@ -153,7 +154,10 @@ public void testReinitSecureCredentials() {
             final MockSecureSettings newSecureSettings = new MockSecureSettings();
             newSecureSettings.setString("s3.client." + clientName + ".access_key", "new_secret_aws_key");
             newSecureSettings.setString("s3.client." + clientName + ".secret_key", "new_secret_aws_secret");
-            final Settings newSettings = Settings.builder().setSecureSettings(newSecureSettings).build();
+            final Settings newSettings = Settings.builder()
+                .put(S3Repository.BUCKET_SETTING.getKey(), "bucket")
+                .setSecureSettings(newSecureSettings)
+                .build();
             // reload S3 plugin settings
             final PluginsService plugins = getInstanceFromNode(PluginsService.class);
             final ProxyS3RepositoryPlugin plugin = plugins.filterPlugins(ProxyS3RepositoryPlugin.class).findFirst().get();
@@ -202,6 +206,7 @@ public void testInsecureRepositoryCredentials() throws Exception {
         createRepository(
             repositoryName,
             Settings.builder()
+                .put(S3Repository.BUCKET_SETTING.getKey(), "bucket")
                 .put(S3Repository.ACCESS_KEY_SETTING.getKey(), "insecure_aws_key")
                 .put(S3Repository.SECRET_KEY_SETTING.getKey(), "insecure_aws_secret")
                 .build()

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

@@ -89,6 +89,7 @@ public void testInvalidChunkBufferSizeSettings() {
 
     private Settings bufferAndChunkSettings(long buffer, long chunk) {
         return Settings.builder()
+            .put(S3Repository.BUCKET_SETTING.getKey(), "bucket")
             .put(S3Repository.BUFFER_SIZE_SETTING.getKey(), new ByteSizeValue(buffer, ByteSizeUnit.MB).getStringRep())
             .put(S3Repository.CHUNK_SIZE_SETTING.getKey(), new ByteSizeValue(chunk, ByteSizeUnit.MB).getStringRep())
             .build();
@@ -102,15 +103,22 @@ public void testBasePathSetting() {
         final RepositoryMetadata metadata = new RepositoryMetadata(
             "dummy-repo",
             "mock",
-            Settings.builder().put(S3Repository.BASE_PATH_SETTING.getKey(), "foo/bar").build()
+            Settings.builder()
+                .put(S3Repository.BUCKET_SETTING.getKey(), "bucket")
+                .put(S3Repository.BASE_PATH_SETTING.getKey(), "foo/bar")
+                .build()
         );
         try (S3Repository s3repo = createS3Repo(metadata)) {
             assertEquals("foo/bar/", s3repo.basePath().buildAsString());
         }
     }
 
     public void testDefaultBufferSize() {
-        final RepositoryMetadata metadata = new RepositoryMetadata("dummy-repo", "mock", Settings.EMPTY);
+        final RepositoryMetadata metadata = new RepositoryMetadata(
+            "dummy-repo",
+            "mock",
+            Settings.builder().put(S3Repository.BUCKET_SETTING.getKey(), "bucket").build()
+        );
         try (S3Repository s3repo = createS3Repo(metadata)) {
             assertThat(s3repo.getBlobStore(), is(nullValue()));
             s3repo.start();
@@ -121,6 +129,17 @@ public void testDefaultBufferSize() {
         }
     }
 
+    public void testMissingBucketName() {
+        final var metadata = new RepositoryMetadata("repo", "mock", Settings.EMPTY);
+        assertThrows(IllegalArgumentException.class, () -> createS3Repo(metadata));
+    }
+
+    public void testEmptyBucketName() {
+        final var settings = Settings.builder().put(S3Repository.BUCKET_SETTING.getKey(), "").build();
+        final var metadata = new RepositoryMetadata("repo", "mock", settings);
+        assertThrows(IllegalArgumentException.class, () -> createS3Repo(metadata));
+    }
+
     private S3Repository createS3Repo(RepositoryMetadata metadata) {
         return new S3Repository(
             metadata,
@@ -132,4 +151,5 @@ private S3Repository createS3Repo(RepositoryMetadata metadata) {
             S3RepositoriesMetrics.NOOP
         );
     }
+
 }

modules/repository-s3/src/yamlRestTest/resources/rest-api-spec/test/repository_s3/10_basic.yml

@@ -1,16 +1,46 @@
 # Integration tests for repository-s3
 #
 "Module repository-s3 is loaded":
-    - skip:
-        reason: "contains is a newly added assertion"
-        features: contains
-    - do:
-        cluster.state: {}
+  - skip:
+      reason: "contains is a newly added assertion"
+      features: contains
+  - do:
+      cluster.state: { }
 
-    # Get master node id
-    - set: { master_node: master }
+  # Get master node id
+  - set: { master_node: master }
 
-    - do:
-        nodes.info: {}
+  - do:
+      nodes.info: { }
 
-    - contains:  { nodes.$master.modules: { name: repository-s3 } }
+  - contains: { nodes.$master.modules: { name: repository-s3 } }
+
+---
+"Create S3 snapshot repository":
+  - do:
+      snapshot.create_repository:
+        repository: test-snapshot-repo
+        verify: false
+        body:
+          type: s3
+          settings:
+            bucket: test-bucket
+
+---
+"Create S3 snapshot repository without bucket":
+  - do:
+      catch: /illegal_argument_exception/
+      snapshot.create_repository:
+        repository: test-snapshot-repo-without-bucket
+        verify: false
+        body:
+          type: s3
+  - do:
+      catch: /illegal_argument_exception/
+      snapshot.create_repository:
+        repository: test-snapshot-repo-without-bucket
+        verify: false
+        body:
+          type: s3
+          settings:
+            bucket: ""