Fix cross cluster search with security (elastic/x-pack-elasticsearch#904)

This commit adds an integration test that runs basic cross cluster search
actions across 2 clusters with security installed. This commit also fixes
several issues with respect to internal actions and proxy actions in the context
of cross cluster search.

Relates to elastic/elasticsearch#23830
relates elastic/x-pack-elasticsearch#892

Original commit: elastic/x-pack-elasticsearch@2e5486c259

plugin/src/main/java/org/elasticsearch/xpack/security/authz/AuthorizationService.java

@@ -32,6 +32,7 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.common.util.set.Sets;
 import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.transport.TransportActionProxy;
 import org.elasticsearch.transport.TransportRequest;
 import org.elasticsearch.xpack.security.SecurityLifecycleService;
 import org.elasticsearch.xpack.security.action.user.AuthenticateAction;
@@ -126,6 +127,7 @@ public class AuthorizationService extends AbstractComponent {
         if (request instanceof ConcreteShardRequest) {
             request = ((ConcreteShardRequest<?>) request).getRequest();
         }
+        request = TransportActionProxy.unwrapRequest(request);
         // prior to doing any authorization lets set the originating action in the context only
         setOriginatingAction(action);
 

plugin/src/main/java/org/elasticsearch/xpack/security/authz/AuthorizationUtils.java

@@ -38,7 +38,7 @@ public final class AuthorizationUtils {
      * @return true if the system user should be used to execute a request
      */
     public static boolean shouldReplaceUserWithSystem(ThreadContext threadContext, String action) {
-        if (isInternalAction(action) == false) {
+        if (threadContext.isSystemContext() == false && isInternalAction(action) == false) {
             return false;
         }
 

plugin/src/main/java/org/elasticsearch/xpack/security/authz/privilege/IndexPrivilege.java

@@ -37,8 +37,9 @@ import static org.elasticsearch.xpack.security.support.Automatons.unionAndMinimi
 
 public final class IndexPrivilege extends Privilege {
 
-    private static final Automaton ALL_AUTOMATON = patterns("indices:*");
+    private static final Automaton ALL_AUTOMATON = patterns("indices:*", "internal:transport/proxy/indices:*");
     private static final Automaton READ_AUTOMATON = patterns("indices:data/read/*");
+    private static final Automaton READ_CROSS_CLUSTER_AUTOMATON = patterns("internal:transport/proxy/indices:data/read/*");
     private static final Automaton CREATE_AUTOMATON = patterns("indices:data/write/index*", "indices:data/write/bulk*",
             PutMappingAction.NAME);
     private static final Automaton INDEX_AUTOMATON =
@@ -57,6 +58,7 @@ public final class IndexPrivilege extends Privilege {
     public static final IndexPrivilege NONE =                new IndexPrivilege("none",             Automatons.EMPTY);
     public static final IndexPrivilege ALL =                 new IndexPrivilege("all",              ALL_AUTOMATON);
     public static final IndexPrivilege READ =                new IndexPrivilege("read",                READ_AUTOMATON);
+    public static final IndexPrivilege READ_CROSS_CLUSTER =  new IndexPrivilege("read_cross_cluster",  READ_CROSS_CLUSTER_AUTOMATON);
     public static final IndexPrivilege CREATE =              new IndexPrivilege("create",              CREATE_AUTOMATON);
     public static final IndexPrivilege INDEX =               new IndexPrivilege("index",               INDEX_AUTOMATON);
     public static final IndexPrivilege DELETE =              new IndexPrivilege("delete",              DELETE_AUTOMATON);
@@ -80,6 +82,7 @@ public final class IndexPrivilege extends Privilege {
             .put("create", CREATE)
             .put("delete_index", DELETE_INDEX)
             .put("view_index_metadata", VIEW_METADATA)
+            .put("read_cross_cluster", READ_CROSS_CLUSTER)
             .immutableMap();
 
     public static final Predicate<String> ACTION_MATCHER = ALL.predicate();

plugin/src/main/java/org/elasticsearch/xpack/security/authz/privilege/SystemPrivilege.java

@@ -14,13 +14,14 @@ public final class SystemPrivilege extends Privilege {
 
     public static SystemPrivilege INSTANCE = new SystemPrivilege();
 
-    private static final Predicate<String> PREDICATE = Automatons.predicate(
+    private static final Predicate<String> PREDICATE = Automatons.predicate(Automatons.
+            minusAndMinimize(Automatons.patterns(
             "internal:*",
             "indices:monitor/*", // added for monitoring
             "cluster:monitor/*",  // added for monitoring
             "cluster:admin/reroute", // added for DiskThresholdDecider.DiskListener
             "indices:admin/mapping/put" // needed for recovery and shrink api
-    );
+    ), Automatons.patterns("internal:transport/proxy/*"))); // no proxy actions for system user!
 
     private SystemPrivilege() {
         super(Collections.singleton("internal"));

plugin/src/test/java/org/elasticsearch/xpack/security/authz/privilege/PrivilegeTests.java

@@ -86,6 +86,8 @@ public class PrivilegeTests extends ESTestCase {
         assertThat(index, notNullValue());
         assertThat(index.predicate().test("indices:admin/mapping/delete"), is(true));
         assertThat(index.predicate().test("indices:admin/mapping/dele"), is(false));
+        assertThat(IndexPrivilege.READ_CROSS_CLUSTER.predicate()
+                .test("internal:transport/proxy/indices:data/read/query"), is(true));
     }
 
     public void testIndexCollapse() throws Exception {
@@ -120,5 +122,6 @@ public class PrivilegeTests extends ESTestCase {
         assertThat(predicate.test("cluster:admin/whatever"), is(false));
         assertThat(predicate.test("indices:admin/mapping/put"), is(true));
         assertThat(predicate.test("indices:admin/mapping/whatever"), is(false));
+        assertThat(predicate.test("internal:transport/proxy/indices:data/read/query"), is(false));
     }
 }

qa/multi-cluster-search-security/build.gradle

@@ -0,0 +1,77 @@
+import org.elasticsearch.gradle.test.RestIntegTestTask
+
+apply plugin: 'elasticsearch.standalone-test'
+
+dependencies {
+  testCompile project(path: ':x-pack-elasticsearch:plugin', configuration: 'runtime')
+}
+
+task remoteClusterTest(type: RestIntegTestTask) {
+  mustRunAfter(precommit)
+}
+
+remoteClusterTestCluster {
+  distribution = 'zip'
+  numNodes = 2
+  clusterName = 'remote-cluster'
+  setting 'search.remote.connect', false
+  plugin ':x-pack-elasticsearch:plugin'
+  setting 'xpack.watcher.enabled', 'false'
+  setting 'xpack.monitoring.enabled', 'false'
+  setting 'xpack.ml.enabled', 'false'
+  setupCommand 'setupDummyUser',
+          'bin/x-pack/users', 'useradd', 'test_user', '-p', 'changeme', '-r', 'superuser'
+  waitCondition = { node, ant ->
+    File tmpFile = new File(node.cwd, 'wait.success')
+    ant.get(src: "http://${node.httpUri()}/_cluster/health?wait_for_nodes=>=${numNodes}&wait_for_status=yellow",
+            dest: tmpFile.toString(),
+            username: 'test_user',
+            password: 'changeme',
+            ignoreerrors: true,
+            retries: 10)
+    return tmpFile.exists()
+  }
+}
+
+remoteClusterTestRunner {
+  systemProperty 'tests.rest.suite', 'remote_cluster'
+}
+
+task mixedClusterTest(type: RestIntegTestTask) {
+  dependsOn(remoteClusterTestRunner)
+}
+
+mixedClusterTestCluster {
+  plugin ':x-pack-elasticsearch:plugin'
+  setting 'xpack.watcher.enabled', 'false'
+  setting 'xpack.monitoring.enabled', 'false'
+  setting 'xpack.ml.enabled', 'false'
+  setupCommand 'setupDummyUser',
+          'bin/x-pack/users', 'useradd', 'test_user', '-p', 'changeme', '-r', 'superuser'
+  waitCondition = { node, ant ->
+    File tmpFile = new File(node.cwd, 'wait.success')
+    ant.get(src: "http://${node.httpUri()}/_cluster/health?wait_for_nodes=>=${numNodes}&wait_for_status=yellow",
+            dest: tmpFile.toString(),
+            username: 'test_user',
+            password: 'changeme',
+            ignoreerrors: true,
+            retries: 10)
+    return tmpFile.exists()
+  }
+  distribution = 'zip'
+  setting 'search.remote.my_remote_cluster.seeds', "\"${-> remoteClusterTest.nodes.get(0).transportUri()}\""
+  setting 'search.remote.connections_per_cluster', 1
+  setting 'search.remote.connect', true
+}
+
+mixedClusterTestRunner {
+  systemProperty 'tests.rest.suite', 'multi_cluster'
+  finalizedBy 'remoteClusterTestCluster#node0.stop','remoteClusterTestCluster#node1.stop'
+}
+
+task integTest {
+  dependsOn = [mixedClusterTest]
+}
+
+test.enabled = false // no unit tests for multi-cluster-search, only the rest integration test
+check.dependsOn(integTest)

qa/multi-cluster-search-security/src/test/java/org/elasticsearch/xpack/security/MultiClusterSearchWithSecurityYamlTestSuiteIT.java

@@ -0,0 +1,49 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.xpack.security;
+
+import com.carrotsearch.randomizedtesting.annotations.Name;
+import com.carrotsearch.randomizedtesting.annotations.ParametersFactory;
+
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.concurrent.ThreadContext;
+import org.elasticsearch.test.rest.yaml.ClientYamlTestCandidate;
+import org.elasticsearch.test.rest.yaml.ESClientYamlSuiteTestCase;
+import org.elasticsearch.xpack.security.authc.support.SecuredString;
+
+import java.io.IOException;
+
+import static org.elasticsearch.xpack.security.authc.support.UsernamePasswordToken.basicAuthHeaderValue;
+
+public class MultiClusterSearchWithSecurityYamlTestSuiteIT extends ESClientYamlSuiteTestCase {
+
+    private static final String USER = "test_user";
+    private static final String PASS = "changeme";
+
+    @Override
+    protected boolean preserveIndicesUponCompletion() {
+        return true;
+    }
+
+    public MultiClusterSearchWithSecurityYamlTestSuiteIT(
+            @Name("yaml") ClientYamlTestCandidate testCandidate) {
+        super(testCandidate);
+    }
+
+    @ParametersFactory
+    public static Iterable<Object[]> parameters() throws IOException {
+        return ESClientYamlSuiteTestCase.createParameters();
+    }
+
+    @Override
+    protected Settings restClientSettings() {
+        String token = basicAuthHeaderValue(USER, new SecuredString(PASS.toCharArray()));
+        return Settings.builder()
+                .put(ThreadContext.PREFIX + ".Authorization", token)
+                .build();
+    }
+}
+

qa/multi-cluster-search-security/src/test/resources/rest-api-spec/test/multi_cluster/10_basic.yaml

@@ -0,0 +1,176 @@
+---
+setup:
+  - skip:
+      features: headers
+
+  - do:
+      cluster.health:
+        wait_for_status: yellow
+  - do:
+      xpack.security.put_user:
+        username: "joe"
+        body:  >
+            {
+              "password": "s3krit",
+              "roles" : [ "x_cluster_role" ]
+            }
+  - do:
+      xpack.security.put_role:
+        name: "x_cluster_role"
+        body:  >
+            {
+              "cluster": ["all"],
+              "indices": [
+                {
+                  "names": "*",
+                  "privileges": ["read", "read_cross_cluster", "view_index_metadata"]
+                }
+              ]
+            }
+---
+teardown:
+  - do:
+      xpack.security.delete_user:
+        username: "joe"
+        ignore: 404
+  - do:
+      xpack.security.delete_role:
+        name: "x_cluster_role"
+        ignore: 404
+---
+"Index data and search on the mixed cluster":
+
+  - do:
+      indices.create:
+        index: test_index
+        body:
+          settings:
+            index:
+              number_of_shards: 2
+              number_of_replicas: 0
+
+  - do:
+      bulk:
+        refresh: true
+        body:
+            - '{"index": {"_index": "test_index", "_type": "test_type"}}'
+            - '{"f1": "local_cluster", "filter_field": 0}'
+            - '{"index": {"_index": "test_index", "_type": "test_type"}}'
+            - '{"f1": "local_cluster", "filter_field": 1}'
+            - '{"index": {"_index": "test_index", "_type": "test_type"}}'
+            - '{"f1": "local_cluster", "filter_field": 0}'
+            - '{"index": {"_index": "test_index", "_type": "test_type"}}'
+            - '{"f1": "local_cluster", "filter_field": 1}'
+            - '{"index": {"_index": "test_index", "_type": "test_type"}}'
+            - '{"f1": "local_cluster", "filter_field": 0}'
+
+  - do:
+      headers: { Authorization: "Basic am9lOnMza3JpdA==" }
+      search:
+        index: test_index,my_remote_cluster:test_index
+        body:
+          aggs:
+            cluster:
+              terms:
+                field: f1.keyword
+
+  - match: { _shards.total: 5 }
+  - match: { hits.total: 11 }
+  - length: { aggregations.cluster.buckets: 2 }
+  - match: { aggregations.cluster.buckets.0.key: "remote_cluster" }
+  - match: { aggregations.cluster.buckets.0.doc_count: 6 }
+  - match: { aggregations.cluster.buckets.1.key: "local_cluster" }
+  - match: { aggregations.cluster.buckets.1.doc_count: 5 }
+
+  - do:
+      headers: { Authorization: "Basic am9lOnMza3JpdA==" }
+      search:
+        index: test_index,my_remote_cluster:test_index
+        body:
+          query:
+            term:
+              f1: remote_cluster
+          aggs:
+            cluster:
+              terms:
+                field: f1.keyword
+
+  - match: { _shards.total: 5 }
+  - match: { hits.total: 6}
+  - match: { hits.hits.0._index: "my_remote_cluster:test_index"}
+  - length: { aggregations.cluster.buckets: 1 }
+  - match: { aggregations.cluster.buckets.0.key: "remote_cluster" }
+  - match: { aggregations.cluster.buckets.0.doc_count: 6 }
+
+  - do:
+      headers: { Authorization: "Basic am9lOnMza3JpdA==" }
+      search:
+        index: my_remote_cluster:test_index
+        body:
+          aggs:
+            cluster:
+              terms:
+                field: f1.keyword
+
+  - match: { _shards.total: 3 }
+  - match: { hits.total: 6}
+  - match: { hits.hits.0._index: "my_remote_cluster:test_index"}
+  - length: { aggregations.cluster.buckets: 1 }
+  - match: { aggregations.cluster.buckets.0.key: "remote_cluster" }
+  - match: { aggregations.cluster.buckets.0.doc_count: 6 }
+
+  - do:
+      headers: { Authorization: "Basic am9lOnMza3JpdA==" }
+      search:
+        index: test_index
+        body:
+          aggs:
+            cluster:
+              terms:
+                field: f1.keyword
+
+  - match: { _shards.total: 2 }
+  - match: { hits.total: 5}
+  - match: { hits.hits.0._index: "test_index"}
+  - length: { aggregations.cluster.buckets: 1 }
+  - match: { aggregations.cluster.buckets.0.key: "local_cluster" }
+  - match: { aggregations.cluster.buckets.0.doc_count: 5 }
+
+---
+"Add transient remote cluster based on the preset cluster":
+  - do:
+      cluster.get_settings:
+        include_defaults: true
+
+  - set: { defaults.search.remote.my_remote_cluster.seeds.0: remote_ip }
+
+  - do:
+      cluster.put_settings:
+        flat_settings: true
+        body:
+          transient:
+            search.remote.test_remote_cluster.seeds: $remote_ip
+
+  - match: {transient: {search.remote.test_remote_cluster.seeds: $remote_ip}}
+
+  - do:
+      headers: { Authorization: "Basic am9lOnMza3JpdA==" }
+      search:
+        index: test_remote_cluster:test_index
+
+  - match: { _shards.total: 3 }
+  - match: { hits.total: 6 }
+  - match: { hits.hits.0._index: "test_remote_cluster:test_index" }
+
+---
+"Search an filtered alias on the remote cluster":
+
+  - do:
+      headers: { Authorization: "Basic am9lOnMza3JpdA==" }
+      search:
+        index: my_remote_cluster:aliased_test_index
+
+  - match: { _shards.total: 3 }
+  - match: { hits.total: 2 }
+  - match: { hits.hits.0._source.filter_field: 1 }
+  - match: { hits.hits.0._index: "my_remote_cluster:test_index" }

qa/multi-cluster-search-security/src/test/resources/rest-api-spec/test/remote_cluster/10_basic.yaml

@@ -0,0 +1,88 @@
+---
+setup:
+  - skip:
+      features: headers
+
+  - do:
+      cluster.health:
+        wait_for_status: yellow
+  - do:
+      xpack.security.put_user:
+        username: "joe"
+        body:  >
+            {
+              "password": "s3krit",
+              "roles" : [ "x_cluster_role" ]
+            }
+  - do:
+      xpack.security.put_role:
+        name: "x_cluster_role"
+        body:  >
+            {
+              "cluster": ["all"],
+              "indices": [
+                {
+                  "names": "*",
+                  "privileges": ["read", "read_cross_cluster", "view_index_metadata"]
+                }
+              ]
+            }
+---
+"Index data and search on the remote cluster":
+
+  - do:
+      indices.create:
+        index: test_index
+        body:
+          settings:
+            index:
+              number_of_shards: 3
+              number_of_replicas: 0
+          aliases:
+            aliased_test_index: # we use this alias in the multi cluster test to verify filtered aliases work
+              filter:
+                term:
+                  filter_field : 1
+
+  - do:
+      bulk:
+         refresh: true
+         body:
+           - '{"index": {"_index": "test_index", "_type": "test_type"}}'
+           - '{"f1": "remote_cluster", "filter_field": 0}'
+           - '{"index": {"_index": "test_index", "_type": "test_type"}}'
+           - '{"f1": "remote_cluster", "filter_field": 1}'
+           - '{"index": {"_index": "test_index", "_type": "test_type"}}'
+           - '{"f1": "remote_cluster", "filter_field": 0}'
+           - '{"index": {"_index": "test_index", "_type": "test_type"}}'
+           - '{"f1": "remote_cluster", "filter_field": 1}'
+           - '{"index": {"_index": "test_index", "_type": "test_type"}}'
+           - '{"f1": "remote_cluster", "filter_field": 0}'
+           - '{"index": {"_index": "test_index", "_type": "test_type"}}'
+           - '{"f1": "remote_cluster", "filter_field": 0}'
+
+  - do:
+      headers: { Authorization: "Basic am9lOnMza3JpdA==" }
+      search:
+        index: test_index
+        body:
+          aggs:
+            cluster:
+              terms:
+                field: f1.keyword
+
+  - match: { _shards.total: 3 }
+  - match: { hits.total: 6 }
+  - length: { aggregations.cluster.buckets: 1 }
+  - match: { aggregations.cluster.buckets.0.key: "remote_cluster" }
+  - match: { aggregations.cluster.buckets.0.doc_count: 6 }
+
+  - do:
+      headers: { Authorization: "Basic am9lOnMza3JpdA==" }
+      search:
+        index: aliased_test_index
+
+  - match: { _shards.total: 3 }
+  - match: { hits.total: 2 }
+  - match: { hits.hits.0._source.filter_field: 1 }
+  - match: { hits.hits.0._index: "test_index" }