Resolve/cluster should mark remotes as not connected when a security …

…exception is thrown (elastic#119793)

Fixes two bugs in _resolve/cluster.

First, the code that detects older clusters versions and does a fallback to the _resolve/index
endpoint was using an outdated string match for error detection. That has been adjusted.

Second, upon security exceptions, the _resolve/cluster endpoint was marking the clusters as connected: true,
under the assumption that all security exceptions related to cross cluster calls and remote index access were
coming from the remote cluster, but that is not always the case. Some cross-cluster security violations can
be detected on the local querying cluster after issuing the remoteClient.execute call but before the transport
layer actually sends the request remotely. So we now mark the connected status as false for all ElasticsearchSecurityException cases. End user docs have been updated with this information.

docs/changelog/119793.yaml

@@ -0,0 +1,6 @@
+pr: 119793
+summary: Resolve/cluster should mark remotes as not connected when a security exception
+  is thrown
+area: CCS
+type: bug
+issues: []

docs/reference/indices/resolve-cluster.asciidoc

@@ -28,8 +28,19 @@ For each cluster in the index expression, information is returned about:
 3. whether there are any indices, aliases or data streams on that cluster that match
    the index expression
 4. whether the search is likely to have errors returned when you do the {ccs} (including any
-   authorization errors if your user does not have permission to query the index)
-5. cluster version information, including the Elasticsearch server version
+   authorization errors if your user does not have permission to query a remote cluster or
+   the indices on that cluster)
+5. (in some cases) cluster version information, including the Elasticsearch server version
+
+[TIP]
+====
+Whenever a security exception is returned for a remote cluster, that remote
+will always be marked as connected=false in the response, since your user does not have
+permissions to access that cluster (or perhaps the remote index) you are querying.
+Once the proper security permissions are obtained, then you can rely on the `connected` field
+in the response to determine whether the remote cluster is available and ready for querying.
+====
+
 
 ////
 [source,console]

server/src/main/java/org/elasticsearch/action/admin/indices/resolve/ResolveClusterActionRequest.java

@@ -9,12 +9,14 @@
 
 package org.elasticsearch.action.admin.indices.resolve;
 
+import org.elasticsearch.TransportVersion;
 import org.elasticsearch.TransportVersions;
 import org.elasticsearch.action.ActionRequest;
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.IndicesRequest;
 import org.elasticsearch.action.ValidateActions;
 import org.elasticsearch.action.support.IndicesOptions;
+import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.tasks.CancellableTask;
@@ -30,6 +32,7 @@
 public class ResolveClusterActionRequest extends ActionRequest implements IndicesRequest.Replaceable {
 
     public static final IndicesOptions DEFAULT_INDICES_OPTIONS = IndicesOptions.strictExpandOpen();
+    public static final String TRANSPORT_VERSION_ERROR_MESSAGE_PREFIX = "ResolveClusterAction requires at least version";
 
     private String[] names;
     /*
@@ -65,12 +68,7 @@ public ResolveClusterActionRequest(String[] names, IndicesOptions indicesOptions
     public ResolveClusterActionRequest(StreamInput in) throws IOException {
         super(in);
         if (in.getTransportVersion().before(TransportVersions.V_8_13_0)) {
-            throw new UnsupportedOperationException(
-                "ResolveClusterAction requires at least version "
-                    + TransportVersions.V_8_13_0.toReleaseVersion()
-                    + " but was "
-                    + in.getTransportVersion().toReleaseVersion()
-            );
+            throw new UnsupportedOperationException(createVersionErrorMessage(in.getTransportVersion()));
         }
         this.names = in.readStringArray();
         this.indicesOptions = IndicesOptions.readIndicesOptions(in);
@@ -81,17 +79,21 @@ public ResolveClusterActionRequest(StreamInput in) throws IOException {
     public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
         if (out.getTransportVersion().before(TransportVersions.V_8_13_0)) {
-            throw new UnsupportedOperationException(
-                "ResolveClusterAction requires at least version "
-                    + TransportVersions.V_8_13_0.toReleaseVersion()
-                    + " but was "
-                    + out.getTransportVersion().toReleaseVersion()
-            );
+            throw new UnsupportedOperationException(createVersionErrorMessage(out.getTransportVersion()));
         }
         out.writeStringArray(names);
         indicesOptions.writeIndicesOptions(out);
     }
 
+    private String createVersionErrorMessage(TransportVersion versionFound) {
+        return Strings.format(
+            "%s %s but was %s",
+            TRANSPORT_VERSION_ERROR_MESSAGE_PREFIX,
+            TransportVersions.V_8_13_0.toReleaseVersion(),
+            versionFound.toReleaseVersion()
+        );
+    }
+
     @Override
     public ActionRequestValidationException validate() {
         ActionRequestValidationException validationException = null;

server/src/main/java/org/elasticsearch/action/admin/indices/resolve/TransportResolveClusterAction.java

@@ -51,7 +51,6 @@
 public class TransportResolveClusterAction extends HandledTransportAction<ResolveClusterActionRequest, ResolveClusterActionResponse> {
 
     private static final Logger logger = LogManager.getLogger(TransportResolveClusterAction.class);
-    private static final String TRANSPORT_VERSION_ERROR_MESSAGE = "ResolveClusterAction requires at least Transport Version";
 
     public static final String NAME = "indices:admin/resolve/cluster";
     public static final ActionType<ResolveClusterActionResponse> TYPE = new ActionType<>(NAME);
@@ -175,7 +174,13 @@ public void onFailure(Exception failure) {
                             failure,
                             ElasticsearchSecurityException.class
                         ) instanceof ElasticsearchSecurityException ese) {
-                            clusterInfoMap.put(clusterAlias, new ResolveClusterInfo(true, skipUnavailable, ese.getMessage()));
+                            /*
+                             * some ElasticsearchSecurityExceptions come from the local cluster security interceptor after you've
+                             * issued the client.execute call but before any call went to the remote cluster, so with an
+                             * ElasticsearchSecurityException you can't tell whether the remote cluster is available or not, so mark
+                             * it as connected=false
+                             */
+                            clusterInfoMap.put(clusterAlias, new ResolveClusterInfo(false, skipUnavailable, ese.getMessage()));
                         } else if (ExceptionsHelper.unwrap(failure, IndexNotFoundException.class) instanceof IndexNotFoundException infe) {
                             clusterInfoMap.put(clusterAlias, new ResolveClusterInfo(true, skipUnavailable, infe.getMessage()));
                         } else {
@@ -184,7 +189,7 @@ public void onFailure(Exception failure) {
                             // this error at the Transport layer BEFORE it sends the request to the remote cluster, since there
                             // are version guards on the Writeables for this Action, namely ResolveClusterActionRequest.writeTo
                             if (cause instanceof UnsupportedOperationException
-                                && cause.getMessage().contains(TRANSPORT_VERSION_ERROR_MESSAGE)) {
+                                && cause.getMessage().contains(ResolveClusterActionRequest.TRANSPORT_VERSION_ERROR_MESSAGE_PREFIX)) {
                                 // Since this cluster does not have _resolve/cluster, we call the _resolve/index
                                 // endpoint to fill in the matching_indices field of the response for that cluster
                                 ResolveIndexAction.Request resolveIndexRequest = new ResolveIndexAction.Request(

x-pack/plugin/security/qa/multi-cluster/src/javaRestTest/java/org/elasticsearch/xpack/remotecluster/RemoteClusterSecurityRCS1ResolveClusterIT.java

@@ -93,7 +93,9 @@ public void testResolveClusterUnderRCS1() throws Exception {
             assertLocalMatching(responseMap);
 
             Map<String, ?> remoteClusterResponse = (Map<String, ?>) responseMap.get("my_remote_cluster");
-            assertThat((Boolean) remoteClusterResponse.get("connected"), equalTo(true));
+            // with security exceptions, the remote should be marked as connected=false, since you can't tell whether a security
+            // exception comes from the local cluster (intercepted) or the remote
+            assertThat((Boolean) remoteClusterResponse.get("connected"), equalTo(false));
             assertThat((String) remoteClusterResponse.get("error"), containsString("unauthorized for user [remote_search_user]"));
 
             // TEST CASE 2: Query cluster -> add user role and user on remote cluster and try resolve again
@@ -171,7 +173,7 @@ public void testResolveClusterUnderRCS1() throws Exception {
             Map<String, Object> responseMap = responseAsMap(response);
             assertThat(responseMap.get(LOCAL_CLUSTER_NAME_REPRESENTATION), nullValue());
             Map<String, ?> remoteClusterResponse = (Map<String, ?>) responseMap.get("my_remote_cluster");
-            assertThat((Boolean) remoteClusterResponse.get("connected"), equalTo(true));
+            assertThat((Boolean) remoteClusterResponse.get("connected"), equalTo(false));
             assertThat((String) remoteClusterResponse.get("error"), containsString("unauthorized for user [remote_search_user]"));
             assertThat((String) remoteClusterResponse.get("error"), containsString("on indices [secretindex]"));
         }
@@ -183,7 +185,7 @@ public void testResolveClusterUnderRCS1() throws Exception {
             Map<String, Object> responseMap = responseAsMap(response);
             assertThat(responseMap.get(LOCAL_CLUSTER_NAME_REPRESENTATION), nullValue());
             Map<String, ?> remoteClusterResponse = (Map<String, ?>) responseMap.get("my_remote_cluster");
-            assertThat((Boolean) remoteClusterResponse.get("connected"), equalTo(true));
+            assertThat((Boolean) remoteClusterResponse.get("connected"), equalTo(false));
             assertThat((String) remoteClusterResponse.get("error"), containsString("unauthorized for user [remote_search_user]"));
             assertThat((String) remoteClusterResponse.get("error"), containsString("on indices [doesnotexist]"));
         }
@@ -195,6 +197,7 @@ public void testResolveClusterUnderRCS1() throws Exception {
             Map<String, Object> responseMap = responseAsMap(response);
             assertThat(responseMap.get(LOCAL_CLUSTER_NAME_REPRESENTATION), nullValue());
             Map<String, ?> remoteClusterResponse = (Map<String, ?>) responseMap.get("my_remote_cluster");
+            // with IndexNotFoundExceptions, we know that error came from the remote cluster, so we can mark the remote as connected=true
             assertThat((Boolean) remoteClusterResponse.get("connected"), equalTo(true));
             assertThat((String) remoteClusterResponse.get("error"), containsString("no such index [index99]"));
         }
@@ -210,7 +213,7 @@ public void testResolveClusterUnderRCS1() throws Exception {
             Map<String, Object> responseMap = responseAsMap(response);
             assertThat(responseMap.get(LOCAL_CLUSTER_NAME_REPRESENTATION), nullValue());
             Map<String, ?> remoteClusterResponse = (Map<String, ?>) responseMap.get("my_remote_cluster");
-            assertThat((Boolean) remoteClusterResponse.get("connected"), equalTo(true));
+            assertThat((Boolean) remoteClusterResponse.get("connected"), equalTo(false));
             assertThat((String) remoteClusterResponse.get("error"), containsString("unauthorized for user [remote_search_user]"));
             assertThat((String) remoteClusterResponse.get("error"), containsString("on indices [secretindex]"));
         }

x-pack/plugin/security/qa/multi-cluster/src/javaRestTest/java/org/elasticsearch/xpack/remotecluster/RemoteClusterSecurityRCS2ResolveClusterIT.java

@@ -180,7 +180,9 @@ public void testResolveCluster() throws Exception {
             assertLocalMatching(responseMap);
 
             Map<String, ?> remoteClusterResponse = (Map<String, ?>) responseMap.get("my_remote_cluster");
-            assertThat((Boolean) remoteClusterResponse.get("connected"), equalTo(true));
+            // with security exceptions, the remote should be marked as connected=false, since you can't tell whether a security
+            // exception comes from the local cluster (intercepted) or the remote
+            assertThat((Boolean) remoteClusterResponse.get("connected"), equalTo(false));
             assertThat((String) remoteClusterResponse.get("error"), containsString("is unauthorized for user"));
             assertThat(
                 (String) remoteClusterResponse.get("error"),
@@ -261,7 +263,7 @@ public void testResolveCluster() throws Exception {
             Map<String, Object> responseMap = responseAsMap(response);
             assertThat(responseMap.get(LOCAL_CLUSTER_NAME_REPRESENTATION), nullValue());
             Map<String, ?> remoteClusterResponse = (Map<String, ?>) responseMap.get("my_remote_cluster");
-            assertThat((Boolean) remoteClusterResponse.get("connected"), equalTo(true));
+            assertThat((Boolean) remoteClusterResponse.get("connected"), equalTo(false));
             assertThat((String) remoteClusterResponse.get("error"), containsString("is unauthorized for user"));
             assertThat((String) remoteClusterResponse.get("error"), containsString("on indices [secretindex]"));
         }
@@ -273,7 +275,7 @@ public void testResolveCluster() throws Exception {
             Map<String, Object> responseMap = responseAsMap(response);
             assertThat(responseMap.get(LOCAL_CLUSTER_NAME_REPRESENTATION), nullValue());
             Map<String, ?> remoteClusterResponse = (Map<String, ?>) responseMap.get("my_remote_cluster");
-            assertThat((Boolean) remoteClusterResponse.get("connected"), equalTo(true));
+            assertThat((Boolean) remoteClusterResponse.get("connected"), equalTo(false));
             assertThat((String) remoteClusterResponse.get("error"), containsString("is unauthorized for user"));
             assertThat((String) remoteClusterResponse.get("error"), containsString("on indices [doesnotexist]"));
         }
@@ -285,6 +287,7 @@ public void testResolveCluster() throws Exception {
             Map<String, Object> responseMap = responseAsMap(response);
             assertThat(responseMap.get(LOCAL_CLUSTER_NAME_REPRESENTATION), nullValue());
             Map<String, ?> remoteClusterResponse = (Map<String, ?>) responseMap.get("my_remote_cluster");
+            // with IndexNotFoundExceptions, we know that error came from the remote cluster, so we can mark the remote as connected=true
             assertThat((Boolean) remoteClusterResponse.get("connected"), equalTo(true));
             assertThat((Boolean) remoteClusterResponse.get("skip_unavailable"), equalTo(false));
             assertThat((String) remoteClusterResponse.get("error"), containsString("no such index [index99]"));
@@ -301,7 +304,7 @@ public void testResolveCluster() throws Exception {
             Map<String, Object> responseMap = responseAsMap(response);
             assertThat(responseMap.get(LOCAL_CLUSTER_NAME_REPRESENTATION), nullValue());
             Map<String, ?> remoteClusterResponse = (Map<String, ?>) responseMap.get("my_remote_cluster");
-            assertThat((Boolean) remoteClusterResponse.get("connected"), equalTo(true));
+            assertThat((Boolean) remoteClusterResponse.get("connected"), equalTo(false));
             assertThat((String) remoteClusterResponse.get("error"), containsString("is unauthorized for user"));
             assertThat((String) remoteClusterResponse.get("error"), containsString("on indices [secretindex]"));
         }