POC: Cross-Project Search

server/src/main/java/org/elasticsearch/RewritableIndicesRequest.java

@@ -0,0 +1,25 @@
+/*
+ * 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", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+package org.elasticsearch;
+
+import org.elasticsearch.action.IndicesRequest;
+import org.elasticsearch.transport.RemoteClusterService;
+
+import java.util.List;
+
+public interface RewritableIndicesRequest extends IndicesRequest {
+    boolean rewritten();
+
+    void rewritten(List<RewrittenIndexExpression> indexExpressions);
+
+    boolean checkRemote(List<RemoteClusterService.RemoteTag> tags);
+
+    record RewrittenIndexExpression(String original, List<String> rewritten) {}
+}

server/src/main/java/org/elasticsearch/action/search/SearchRequest.java

@@ -9,6 +9,7 @@
 
 package org.elasticsearch.action.search;
 
+import org.elasticsearch.RewritableIndicesRequest;
 import org.elasticsearch.TransportVersions;
 import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionRequestValidationException;
@@ -31,6 +32,7 @@
 import org.elasticsearch.search.sort.SortBuilder;
 import org.elasticsearch.search.sort.SortBuilders;
 import org.elasticsearch.tasks.TaskId;
+import org.elasticsearch.transport.RemoteClusterService;
 import org.elasticsearch.xcontent.ToXContent;
 
 import java.io.IOException;
@@ -53,7 +55,11 @@
  * @see Client#search(SearchRequest)
  * @see SearchResponse
  */
-public class SearchRequest extends LegacyActionRequest implements IndicesRequest.Replaceable, Rewriteable<SearchRequest> {
+public class SearchRequest extends LegacyActionRequest
+    implements
+        RewritableIndicesRequest,
+        IndicesRequest.Replaceable,
+        Rewriteable<SearchRequest> {
 
     public static final ToXContent.Params FORMAT_PARAMS = new ToXContent.MapParams(Collections.singletonMap("pretty", "false"));
 
@@ -69,6 +75,10 @@ public class SearchRequest extends LegacyActionRequest implements IndicesRequest
     private SearchType searchType = SearchType.DEFAULT;
 
     private String[] indices = Strings.EMPTY_ARRAY;
+    private List<RemoteClusterService.RemoteTag> routingTags = List.of();
+
+    @Nullable
+    private List<RewrittenIndexExpression> indexExpressions;
 
     @Nullable
     private String routing;
@@ -400,6 +410,11 @@ public SearchRequest indices(String... indices) {
         return this;
     }
 
+    public SearchRequest routingTags(List<RemoteClusterService.RemoteTag> routingTags) {
+        this.routingTags = routingTags;
+        return this;
+    }
+
     private static void validateIndices(String... indices) {
         Objects.requireNonNull(indices, "indices must not be null");
         for (String index : indices) {
@@ -853,4 +868,30 @@ public String toString() {
             + source
             + '}';
     }
+
+    @Override
+    public boolean rewritten() {
+        return indexExpressions != null;
+    }
+
+    @Override
+    public void rewritten(List<RewrittenIndexExpression> indexExpressions) {
+        assert false == rewritten();
+        this.indexExpressions = indexExpressions;
+        indices(indexExpressions.stream().flatMap(indexExpression -> indexExpression.rewritten().stream()).toArray(String[]::new));
+    }
+
+    @Override
+    public boolean checkRemote(List<RemoteClusterService.RemoteTag> tags) {
+        if (routingTags.isEmpty()) {
+            return true; // no routing requested, so no constraints
+        }
+        // if any tag in routingTags matches one in tags, return true
+        for (RemoteClusterService.RemoteTag tag : routingTags) {
+            if (tags.contains(tag)) {
+                return true;
+            }
+        }
+        return false;
+    }
 }

server/src/main/java/org/elasticsearch/common/settings/ClusterSettings.java

@@ -367,6 +367,7 @@ public void apply(Settings value, Settings current, Settings previous) {
         TransportSearchAction.SHARD_COUNT_LIMIT_SETTING,
         TransportSearchAction.DEFAULT_PRE_FILTER_SHARD_SIZE,
         RemoteClusterService.REMOTE_CLUSTER_SKIP_UNAVAILABLE,
+        RemoteClusterService.REMOTE_CLUSTER_TAGS,
         SniffConnectionStrategy.REMOTE_CONNECTIONS_PER_CLUSTER,
         RemoteClusterService.REMOTE_INITIAL_CONNECTION_TIMEOUT_SETTING,
         RemoteClusterService.REMOTE_NODE_ATTRIBUTE,

server/src/main/java/org/elasticsearch/rest/action/search/RestSearchAction.java

@@ -9,6 +9,8 @@
 
 package org.elasticsearch.rest.action.search;
 
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.search.SearchRequest;
@@ -35,6 +37,7 @@
 import org.elasticsearch.search.sort.SortOrder;
 import org.elasticsearch.search.suggest.SuggestBuilder;
 import org.elasticsearch.search.suggest.term.TermSuggestionBuilder;
+import org.elasticsearch.transport.RemoteClusterService;
 import org.elasticsearch.usage.SearchUsageHolder;
 import org.elasticsearch.xcontent.XContentParser;
 
@@ -63,6 +66,7 @@ public class RestSearchAction extends BaseRestHandler {
     public static final String TYPED_KEYS_PARAM = "typed_keys";
     public static final String INCLUDE_NAMED_QUERIES_SCORE_PARAM = "include_named_queries_score";
     public static final Set<String> RESPONSE_PARAMS = Set.of(TYPED_KEYS_PARAM, TOTAL_HITS_AS_INT_PARAM, INCLUDE_NAMED_QUERIES_SCORE_PARAM);
+    private static final Logger log = LogManager.getLogger(RestSearchAction.class);
 
     private final SearchUsageHolder searchUsageHolder;
     private final Predicate<NodeFeature> clusterSupportsFeature;
@@ -98,6 +102,7 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
             client.threadPool().getThreadContext().setErrorTraceTransportHeader(request);
         }
         SearchRequest searchRequest = new SearchRequest();
+
         // access the BwC param, but just drop it
         // this might be set by old clients
         request.param("min_compatible_shard_node");
@@ -167,6 +172,16 @@ public static void parseSearchRequest(
             searchRequest.source(new SearchSourceBuilder());
         }
         searchRequest.indices(Strings.splitStringByCommaToArray(request.param("index")));
+
+        var routingTags = request.param("query_routing", null);
+        if (routingTags != null) {
+            searchRequest.routingTags(
+                Arrays.stream(Strings.splitStringByCommaToArray(routingTags)).map(RemoteClusterService.RemoteTag::fromString).toList()
+            );
+        } else {
+            log.info("No routing tags");
+        }
+
         if (requestContentParser != null) {
             if (searchUsageHolder == null) {
                 searchRequest.source().parseXContent(requestContentParser, true, clusterSupportsFeature);

server/src/main/java/org/elasticsearch/transport/RemoteClusterService.java

@@ -41,6 +41,7 @@
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
@@ -97,6 +98,33 @@ public final class RemoteClusterService extends RemoteClusterAware
         (ns, key) -> boolSetting(key, true, new RemoteConnectionEnabled<>(ns, key), Setting.Property.Dynamic, Setting.Property.NodeScope)
     );
 
+    public record RemoteTag(String key, String value) {
+        public static RemoteTag fromString(String tag) {
+            if (tag == null || tag.isEmpty()) {
+                throw new IllegalArgumentException("Remote tag must not be null or empty");
+            }
+            // - as a separator to simplify search path param parsing; won't be like this in the real implementation
+            int idx = tag.indexOf('-');
+            if (idx < 0) {
+                return new RemoteTag(tag, "");
+            } else {
+                return new RemoteTag(tag.substring(0, idx), tag.substring(idx + 1));
+            }
+        }
+    }
+
+    public static final Setting.AffixSetting<List<RemoteTag>> REMOTE_CLUSTER_TAGS = Setting.affixKeySetting(
+        "cluster.remote.",
+        "tags",
+        (ns, key) -> Setting.listSetting(
+            key,
+            Collections.emptyList(),
+            RemoteTag::fromString,
+            Setting.Property.Dynamic,
+            Setting.Property.NodeScope
+        )
+    );
+
     public static final Setting.AffixSetting<TimeValue> REMOTE_CLUSTER_PING_SCHEDULE = Setting.affixKeySetting(
         "cluster.remote.",
         "transport.ping_schedule",

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

@@ -0,0 +1,56 @@
+/*
+ * 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;
+
+import org.elasticsearch.transport.Transport;
+import org.elasticsearch.transport.TransportInterceptor;
+import org.elasticsearch.transport.TransportRequest;
+import org.elasticsearch.transport.TransportRequestOptions;
+import org.elasticsearch.transport.TransportResponse;
+import org.elasticsearch.transport.TransportResponseHandler;
+
+public interface CustomRemoteServerTransportInterceptor {
+    // TODO probably don't want this
+    boolean enabled();
+
+    // TODO this should be a wrapper around TransportInterceptor.AsyncSender instead
+    <T extends TransportResponse> void sendRequest(
+        TransportInterceptor.AsyncSender sender,
+        Transport.Connection connection,
+        String action,
+        TransportRequest request,
+        TransportRequestOptions options,
+        TransportResponseHandler<T> handler
+    );
+
+    CustomServerTransportFilter getFilter();
+
+    class Default implements CustomRemoteServerTransportInterceptor {
+        @Override
+        public boolean enabled() {
+            return false;
+        }
+
+        @Override
+        public <T extends TransportResponse> void sendRequest(
+            TransportInterceptor.AsyncSender sender,
+            Transport.Connection connection,
+            String action,
+            TransportRequest request,
+            TransportRequestOptions options,
+            TransportResponseHandler<T> handler
+        ) {
+            sender.sendRequest(connection, action, request, options, handler);
+        }
+
+        @Override
+        public CustomServerTransportFilter getFilter() {
+            return new CustomServerTransportFilter.Default();
+        }
+    }
+}

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

@@ -0,0 +1,22 @@
+/*
+ * 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;
+
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.transport.TransportRequest;
+
+public interface CustomServerTransportFilter {
+    void filter(String securityAction, TransportRequest request, ActionListener<Void> authenticationListener);
+
+    class Default implements CustomServerTransportFilter {
+        @Override
+        public void filter(String securityAction, TransportRequest request, ActionListener<Void> listener) {
+            listener.onResponse(null);
+        }
+    }
+}

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

@@ -21,6 +21,7 @@
 import org.elasticsearch.xpack.core.security.authc.service.ServiceAccountTokenStore;
 import org.elasticsearch.xpack.core.security.authc.support.UserRoleMapper;
 import org.elasticsearch.xpack.core.security.authz.AuthorizationEngine;
+import org.elasticsearch.xpack.core.security.authz.CustomIndicesRequestRewriter;
 import org.elasticsearch.xpack.core.security.authz.RoleDescriptor;
 import org.elasticsearch.xpack.core.security.authz.store.RoleRetrievalResult;
 
@@ -133,6 +134,14 @@ default CustomApiKeyAuthenticator getCustomApiKeyAuthenticator(SecurityComponent
         return null;
     }
 
+    default CustomIndicesRequestRewriter getCustomIndicesRequestRewriter(SecurityComponents components) {
+        return null;
+    }
+
+    default CustomRemoteServerTransportInterceptor getCustomRemoteServerTransportInterceptor(SecurityComponents components) {
+        return null;
+    }
+
     /**
      * Returns a authorization engine for authorizing requests, or null to use the default authorization mechanism.
      *

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authz/CustomIndicesRequestRewriter.java

@@ -0,0 +1,22 @@
+/*
+ * 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.authz;
+
+import org.elasticsearch.RewritableIndicesRequest;
+
+public interface CustomIndicesRequestRewriter {
+    void rewrite(RewritableIndicesRequest request);
+
+    class Default implements CustomIndicesRequestRewriter {
+        @Override
+        public void rewrite(RewritableIndicesRequest request) {
+            // No rewriting by default
+            // This is a no-op implementation
+        }
+    }
+}

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/core/security/transport/netty4/SecurityNetty4Transport.java

@@ -161,7 +161,7 @@ protected ChannelHandler getClientChannelInitializer(DiscoveryNode node, Connect
 
     @Override
     protected InboundPipeline getInboundPipeline(Channel channel, boolean isRemoteClusterServerChannel) {
-        if (false == isRemoteClusterServerChannel) {
+        if (false == isRemoteClusterServerChannel || crossClusterAccessAuthenticationService.skipTransportCheck()) {
             return super.getInboundPipeline(channel, false);
         } else {
             return new InboundPipeline(