[CCR] Enforce auto follow pattern name restrictions (#35197)

An auto follow pattern:
* cannot start with `_`
* cannot contain a `,`
* can be encoded in UTF-8
* the length of UTF-8 encoded bytes is no longer than 255 bytes

Author: martijnvg

x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/action/PutAutoFollowPatternRequestTests.java

@@ -123,4 +123,65 @@ public void testValidate() {
         validationException = request.validate();
         assertThat(validationException, nullValue());
     }
+
+    public void testValidateName() {
+        PutAutoFollowPatternAction.Request request = new PutAutoFollowPatternAction.Request();
+        request.setRemoteCluster("_alias");
+        request.setLeaderIndexPatterns(Collections.singletonList("logs-*"));
+
+        request.setName("name");
+        ActionRequestValidationException validationException = request.validate();
+        assertThat(validationException, nullValue());
+    }
+
+    public void testValidateNameComma() {
+        PutAutoFollowPatternAction.Request request = new PutAutoFollowPatternAction.Request();
+        request.setRemoteCluster("_alias");
+        request.setLeaderIndexPatterns(Collections.singletonList("logs-*"));
+
+        request.setName("name1,name2");
+        ActionRequestValidationException validationException = request.validate();
+        assertThat(validationException, notNullValue());
+        assertThat(validationException.getMessage(), containsString("name must not contain a ','"));
+    }
+
+    public void testValidateNameLeadingUnderscore() {
+        PutAutoFollowPatternAction.Request request = new PutAutoFollowPatternAction.Request();
+        request.setRemoteCluster("_alias");
+        request.setLeaderIndexPatterns(Collections.singletonList("logs-*"));
+
+        request.setName("_name");
+        ActionRequestValidationException validationException = request.validate();
+        assertThat(validationException, notNullValue());
+        assertThat(validationException.getMessage(), containsString("name must not start with '_'"));
+    }
+
+    public void testValidateNameUnderscores() {
+        PutAutoFollowPatternAction.Request request = new PutAutoFollowPatternAction.Request();
+        request.setRemoteCluster("_alias");
+        request.setLeaderIndexPatterns(Collections.singletonList("logs-*"));
+
+        request.setName("n_a_m_e_");
+        ActionRequestValidationException validationException = request.validate();
+        assertThat(validationException, nullValue());
+    }
+
+    public void testValidateNameTooLong() {
+        PutAutoFollowPatternAction.Request request = new PutAutoFollowPatternAction.Request();
+        request.setRemoteCluster("_alias");
+        request.setLeaderIndexPatterns(Collections.singletonList("logs-*"));
+
+        StringBuilder stringBuilder = new StringBuilder();
+        for (int i = 0; i < 256; i++) {
+            stringBuilder.append('x');
+        }
+        request.setName(stringBuilder.toString());
+        ActionRequestValidationException validationException = request.validate();
+        assertThat(validationException, notNullValue());
+        assertThat(validationException.getMessage(), containsString("name is too long (256 > 255)"));
+
+        request.setName("name");
+        validationException = request.validate();
+        assertThat(validationException, nullValue());
+    }
 }

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/ccr/action/PutAutoFollowPatternAction.java

@@ -21,6 +21,7 @@
 import org.elasticsearch.xpack.core.ccr.AutoFollowMetadata.AutoFollowPattern;
 
 import java.io.IOException;
+import java.nio.charset.StandardCharsets;
 import java.util.List;
 import java.util.Objects;
 
@@ -44,8 +45,8 @@ public AcknowledgedResponse newResponse() {
     public static class Request extends AcknowledgedRequest<Request> implements ToXContentObject {
 
         private static final ObjectParser<Request, String> PARSER = new ObjectParser<>("put_auto_follow_pattern_request", Request::new);
-
         private static final ParseField NAME_FIELD = new ParseField("name");
+        private static final int MAX_NAME_BYTES = 255;
 
         static {
             PARSER.declareString(Request::setName, NAME_FIELD);
@@ -119,6 +120,21 @@ public ActionRequestValidationException validate() {
             if (name == null) {
                 validationException = addValidationError("[" + NAME_FIELD.getPreferredName() + "] is missing", validationException);
             }
+            if (name != null) {
+                if (name.contains(",")) {
+                    validationException = addValidationError("[" + NAME_FIELD.getPreferredName() + "] name must not contain a ','",
+                        validationException);
+                }
+                if (name.startsWith("_")) {
+                    validationException = addValidationError("[" + NAME_FIELD.getPreferredName() + "] name must not start with '_'",
+                        validationException);
+                }
+                int byteCount = name.getBytes(StandardCharsets.UTF_8).length;
+                if (byteCount > MAX_NAME_BYTES) {
+                    validationException = addValidationError("[" + NAME_FIELD.getPreferredName() + "] name is too long (" +
+                        byteCount + " > " + MAX_NAME_BYTES + ")", validationException);
+                }
+            }
             if (remoteCluster == null) {
                 validationException = addValidationError("[" + REMOTE_CLUSTER_FIELD.getPreferredName() +
                     "] is missing", validationException);