Fix privilege requirement for CCS with Point In Time reader (#62261) (#62696)

When target indices are remote only, CCS does not require user to have privileges on the local cluster. This PR ensure Point-In-Time reader follows the same pattern.

Relates: #61827

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

@@ -31,6 +31,7 @@ import org.elasticsearch.protocol.xpack.graph.GraphExploreRequest;
 import org.elasticsearch.transport.RemoteClusterAware;
 import org.elasticsearch.transport.RemoteConnectionStrategy;
 import org.elasticsearch.transport.TransportRequest;
+import org.elasticsearch.xpack.core.search.action.OpenPointInTimeRequest;
 import org.elasticsearch.xpack.core.security.authz.ResolvedIndices;
 
 import java.util.ArrayList;
@@ -281,7 +282,8 @@ class IndicesAndAliasesResolver {
 
     static boolean allowsRemoteIndices(IndicesRequest request) {
         return request instanceof SearchRequest || request instanceof FieldCapabilitiesRequest
-                || request instanceof GraphExploreRequest || request instanceof ResolveIndexAction.Request;
+                || request instanceof GraphExploreRequest || request instanceof ResolveIndexAction.Request
+                || request instanceof OpenPointInTimeRequest;
     }
 
     private List<String> loadAuthorizedAliases(List<String> authorizedIndices, Metadata metadata) {

x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authz/AuthorizationServiceTests.java

@@ -86,6 +86,7 @@ import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.common.util.concurrent.ThreadContext.StoredContext;
 import org.elasticsearch.common.xcontent.XContentBuilder;
@@ -97,6 +98,10 @@ import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportActionProxy;
 import org.elasticsearch.transport.TransportRequest;
+import org.elasticsearch.xpack.core.search.action.ClosePointInTimeAction;
+import org.elasticsearch.xpack.core.search.action.ClosePointInTimeRequest;
+import org.elasticsearch.xpack.core.search.action.OpenPointInTimeAction;
+import org.elasticsearch.xpack.core.search.action.OpenPointInTimeRequest;
 import org.elasticsearch.xpack.core.security.action.privilege.DeletePrivilegesAction;
 import org.elasticsearch.xpack.core.security.action.privilege.DeletePrivilegesRequest;
 import org.elasticsearch.xpack.core.security.action.user.AuthenticateAction;
@@ -563,6 +568,51 @@ public class AuthorizationServiceTests extends ESTestCase {
         verifyNoMoreInteractions(auditTrail);
     }
 
+    public void testUserWithNoRolesOpenPointInTimeWithRemoteIndices() {
+        final Authentication authentication = createAuthentication(new User("test user"));
+        mockEmptyMetadata();
+        final String requestId = AuditUtil.getOrGenerateRequestId(threadContext);
+        for (final boolean hasLocalIndices: org.elasticsearch.common.collect.List.of(true, false)) {
+            final String[] indices = new String[] {
+                hasLocalIndices ?
+                    randomAlphaOfLength(5) :
+                    "other_cluster:" + randomFrom(randomAlphaOfLength(5), "*", randomAlphaOfLength(4) + "*"),
+                "other_cluster:" + randomFrom(randomAlphaOfLength(5), "*", randomAlphaOfLength(4) + "*")
+            };
+            final OpenPointInTimeRequest openPointInTimeRequest = new OpenPointInTimeRequest(
+                indices, OpenPointInTimeRequest.DEFAULT_INDICES_OPTIONS, TimeValue.timeValueMinutes(randomLongBetween(1, 10)),
+                randomAlphaOfLength(5), randomAlphaOfLength(5)
+            );
+            if (hasLocalIndices) {
+                assertThrowsAuthorizationException(
+                    () -> authorize(authentication, OpenPointInTimeAction.NAME, openPointInTimeRequest),
+                    "indices:data/read/open_point_in_time", "test user"
+                );
+                verify(auditTrail).accessDenied(eq(requestId), eq(authentication),
+                                                eq("indices:data/read/open_point_in_time"), eq(openPointInTimeRequest),
+                                                authzInfoRoles(Role.EMPTY.names()));
+            } else {
+                authorize(authentication, OpenPointInTimeAction.NAME, openPointInTimeRequest);
+                verify(auditTrail).accessGranted(eq(requestId), eq(authentication),
+                                                 eq("indices:data/read/open_point_in_time"), eq(openPointInTimeRequest),
+                                                 authzInfoRoles(Role.EMPTY.names()));
+            }
+            verifyNoMoreInteractions(auditTrail);
+        }
+    }
+
+    public void testUserWithNoRolesCanClosePointInTime() {
+        final ClosePointInTimeRequest closePointInTimeRequest = new ClosePointInTimeRequest(randomAlphaOfLength(8));
+        final Authentication authentication = createAuthentication(new User("test user"));
+        mockEmptyMetadata();
+        final String requestId = AuditUtil.getOrGenerateRequestId(threadContext);
+        authorize(authentication, ClosePointInTimeAction.NAME, closePointInTimeRequest);
+        verify(auditTrail).accessGranted(eq(requestId), eq(authentication),
+                                         eq("indices:data/read/close_point_in_time"), eq(closePointInTimeRequest),
+                                         authzInfoRoles(Role.EMPTY.names()));
+        verifyNoMoreInteractions(auditTrail);
+    }
+
     public void testUnknownRoleCausesDenial() throws IOException {
         Tuple<String, TransportRequest> tuple = randomFrom(asList(
             new Tuple<>(SearchAction.NAME, new SearchRequest()),

x-pack/qa/multi-cluster-search-security/src/test/resources/rest-api-spec/test/multi_cluster/80_point_in_time.yml

@@ -111,3 +111,39 @@ teardown:
       close_point_in_time:
         body:
           id: "$pit_id"
+
+---
+"Point in time CCS with only remote indices requires no privileges on local cluster":
+
+  - do:
+      headers: { Authorization: "Basic cmVtb3RlOnMza3JpdA==" }
+      open_point_in_time:
+        index: "my_*:point_in_time_index"
+        keep_alive: 5m
+  - set: {id: pit_id}
+
+  - do:
+      headers: { Authorization: "Basic cmVtb3RlOnMza3JpdA==" }
+      search:
+        rest_total_hits_as_int: true
+        sort: created_at
+        body:
+          query:
+            range:
+              created_at:
+                gte: "2020-01-03"
+          pit:
+            id: "$pit_id"
+            keep_alive: 1m
+
+  - match: { hits.total: 2 }
+  - match: { hits.hits.0._index: "my_remote_cluster:point_in_time_index" }
+  - match: { hits.hits.0._source.f: "r3" }
+  - match: { hits.hits.1._index: "my_remote_cluster:point_in_time_index" }
+  - match: { hits.hits.1._source.f: "r4" }
+
+  - do:
+      headers: { Authorization: "Basic cmVtb3RlOnMza3JpdA==" }
+      close_point_in_time:
+        body:
+          id: "$pit_id"