Test to ensure that remote CCS and remote CCR actions don't overlap (#108590)

This commit introduces a test to help ensure that actions used between CCR and CCS 
are evaluated on case-by-case basis if/when the allowed permissions overlap. 
Some cases are fine if they overlap, other cases it could be a security bug. 
This will test will fail the permissions we assign to the RCS 2.0 API key overlap.

x-pack/plugin/core/build.gradle

@@ -180,3 +180,16 @@ if (BuildParams.inFipsJvm) {
   // Test clusters run with security disabled
   tasks.named("javaRestTest").configure { enabled = false }
 }
+
+//this specific test requires a test only system property to be set, so we run it in a different JVM via a separate task
+tasks.register('testAutomatonPatterns', Test) {
+  include '**/AutomatonPatternsTests.class'
+  systemProperty 'tests.automaton.record.patterns', 'true'
+  testClassesDirs = sourceSets.test.output.classesDirs
+  classpath = sourceSets.test.runtimeClasspath
+}
+
+tasks.named('test').configure {
+  exclude '**/AutomatonPatternsTests.class'
+  dependsOn testAutomatonPatterns //to ensure testAutomatonPatterns are run with the test task
+}

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/support/Automatons.java

@@ -23,12 +23,16 @@ import org.elasticsearch.core.TimeValue;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Locale;
+import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ExecutionException;
 import java.util.function.Function;
 import java.util.function.Predicate;
+import java.util.stream.Collectors;
 
 import static org.apache.lucene.util.automaton.Operations.DEFAULT_DETERMINIZE_WORK_LIMIT;
 import static org.apache.lucene.util.automaton.Operations.concatenate;
@@ -69,6 +73,10 @@ public final class Automatons {
     static final char WILDCARD_CHAR = '?';       // Char equality with support for wildcards
     static final char WILDCARD_ESCAPE = '\\';    // Escape character
 
+    // for testing only -Dtests.jvm.argline="-Dtests.automaton.record.patterns=true"
+    public static boolean recordPatterns = System.getProperty("tests.automaton.record.patterns", "false").equals("true");
+    private static final Map<Automaton, List<String>> patternsMap = new HashMap<>();
+
     private Automatons() {}
 
     /**
@@ -87,10 +95,13 @@ public final class Automatons {
             return EMPTY;
         }
         if (cache == null) {
-            return buildAutomaton(patterns);
+            return maybeRecordPatterns(buildAutomaton(patterns), patterns);
         } else {
             try {
-                return cache.computeIfAbsent(Sets.newHashSet(patterns), p -> buildAutomaton((Set<String>) p));
+                return cache.computeIfAbsent(
+                    Sets.newHashSet(patterns),
+                    p -> maybeRecordPatterns(buildAutomaton((Set<String>) p), patterns)
+                );
             } catch (ExecutionException e) {
                 throw unwrapCacheException(e);
             }
@@ -338,4 +349,23 @@ public final class Automatons {
         settingsList.add(CACHE_SIZE);
         settingsList.add(CACHE_TTL);
     }
+
+    private static Automaton maybeRecordPatterns(Automaton automaton, Collection<String> patterns) {
+        if (recordPatterns) {
+            patternsMap.put(
+                automaton,
+                patterns.stream().map(String::trim).map(s -> s.toLowerCase(Locale.ROOT)).sorted().collect(Collectors.toList())
+            );
+        }
+        return automaton;
+    }
+
+    // test only
+    static List<String> getPatterns(Automaton automaton) {
+        if (recordPatterns) {
+            return patternsMap.get(automaton);
+        } else {
+            throw new IllegalArgumentException("recordPatterns is set to false");
+        }
+    }
 }

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/security/authz/privilege/IndexPrivilegeTests.java

@@ -144,4 +144,5 @@ public class IndexPrivilegeTests extends ESTestCase {
         );
         assertThat(Operations.subsetOf(crossClusterReplicationInternal.automaton, IndexPrivilege.get(Set.of("all")).automaton), is(true));
     }
+
 }

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/security/support/AutomatonPatternsTests.java

@@ -0,0 +1,81 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.core.security.support;
+
+import org.apache.lucene.util.automaton.Automaton;
+import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.xpack.core.security.authz.privilege.IndexPrivilege;
+
+import java.util.Arrays;
+import java.util.Locale;
+import java.util.Set;
+
+import static org.elasticsearch.xpack.core.security.action.apikey.CrossClusterApiKeyRoleDescriptorBuilder.CCR_INDICES_PRIVILEGE_NAMES;
+import static org.elasticsearch.xpack.core.security.action.apikey.CrossClusterApiKeyRoleDescriptorBuilder.CCS_INDICES_PRIVILEGE_NAMES;
+
+public class AutomatonPatternsTests extends ESTestCase {
+
+    /**
+     * RCS 2.0 allows a single API key to define "replication" and "search" blocks. If both are defined, this results in an API key with 2
+     * sets of indices permissions. Due to the way API keys (and roles) work across the multiple index permission, the set of index
+     * patterns allowed are effectively the most generous of the sets of index patterns since the index patterns are OR'ed together. For
+     * example, `foo` OR `*` results in access to `*`. So, if you have "search" access defined as `foo`, but replication access defined
+     * as `*`, the API key effectively allows access to index pattern `*`. This means that the access for API keys that define both
+     * "search" and "replication", the action names used are the primary means by which we can constrain CCS to the set of "search" indices
+     * as well as how we constrain CCR to the set "replication" indices. For example, if "replication" ever allowed access to
+     * `indices:data/read/get` for `*` , then the "replication" permissions would effectively enable users of CCS to get documents,
+     * even if "search" is never defined in the RCS 2.0 API key. This obviously is not desirable and in practice when both "search" and
+     * "replication" are defined the isolation between CCS and CCR is only achieved because the action names for the workflows do not
+     * overlap. This test helps to ensure that the actions names used for RCS 2.0 do not bleed over between search and replication.
+     */
+    public void testRemoteClusterPrivsDoNotOverlap() {
+
+        // check that the action patterns for remote CCS are not allowed by remote CCR privileges
+        Arrays.stream(CCS_INDICES_PRIVILEGE_NAMES).forEach(ccsPrivilege -> {
+            Automaton ccsAutomaton = IndexPrivilege.get(Set.of(ccsPrivilege)).getAutomaton();
+            Automatons.getPatterns(ccsAutomaton).forEach(ccsPattern -> {
+                // emulate an action name that could be allowed by a CCS privilege
+                String actionName = ccsPattern.replaceAll("\\*", randomAlphaOfLengthBetween(1, 8));
+                Arrays.stream(CCR_INDICES_PRIVILEGE_NAMES).forEach(ccrPrivileges -> {
+                    String errorMessage = String.format(
+                        Locale.ROOT,
+                        "CCR privilege \"%s\" allows CCS action \"%s\". This could result in an "
+                            + "accidental bleeding of permission between RCS 2.0's search and replication index permissions",
+                        ccrPrivileges,
+                        ccsPattern
+                    );
+                    assertFalse(errorMessage, IndexPrivilege.get(Set.of(ccrPrivileges)).predicate().test(actionName));
+                });
+            });
+        });
+
+        // check that the action patterns for remote CCR are not allowed by remote CCS privileges
+        Arrays.stream(CCR_INDICES_PRIVILEGE_NAMES).forEach(ccrPrivilege -> {
+            Automaton ccrAutomaton = IndexPrivilege.get(Set.of(ccrPrivilege)).getAutomaton();
+            Automatons.getPatterns(ccrAutomaton).forEach(ccrPattern -> {
+                // emulate an action name that could be allowed by a CCR privilege
+                String actionName = ccrPattern.replaceAll("\\*", randomAlphaOfLengthBetween(1, 8));
+                Arrays.stream(CCS_INDICES_PRIVILEGE_NAMES).forEach(ccsPrivileges -> {
+                    if ("indices:data/read/xpack/ccr/shard_changes*".equals(ccrPattern)) {
+                        // do nothing, this action is only applicable to CCR workflows and is a moot point if CCS technically has
+                        // access to the index pattern for this action granted by CCR
+                    } else {
+                        String errorMessage = String.format(
+                            Locale.ROOT,
+                            "CCS privilege \"%s\" allows CCR action \"%s\". This could result in an accidental bleeding of "
+                                + "permission between RCS 2.0's search and replication index permissions",
+                            ccsPrivileges,
+                            ccrPattern
+                        );
+                        assertFalse(errorMessage, IndexPrivilege.get(Set.of(ccsPrivileges)).predicate().test(actionName));
+                    }
+                });
+            });
+        });
+    }
+}