Relax the index access control check for scroll searches (#61446)

The check introduced by #60640 for scroll searches, in which we log
if the index access control before the query and fetch phases differs
from when the scroll context is created, is too strict, leading to spurious
warning log messages.
The check verifies instance equality but this assumes that the fetch
phase is executed in the same thread context as the scroll context
validation. However, this is not true if the scroll search is executed
cross-cluster, and even for local scroll searches it is an unfounded assumption.

The check is hence reduced to a null check for the index access.
The fact that the access control is suitable given the indices that
are actually accessed (by the scroll) will be done in a follow-up,
after we better regulate the creation of index access controls in general.

Author: albertzaharovits

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

@@ -29,7 +29,7 @@
 
 /**
  * A {@link SearchOperationListener} that is used to provide authorization for scroll requests.
- *
+ * <p>
  * In order to identify the user associated with a scroll request, we replace the {@link ReaderContext}
  * on creation with a custom implementation that holds the {@link Authentication} object. When
  * this context is accessed again in {@link SearchOperationListener#onPreQueryPhase(SearchContext)}
@@ -82,7 +82,7 @@ public void validateSearchContext(ReaderContext readerContext, TransportRequest
                 if (null == securityContext.getThreadContext().getTransient(AuthorizationServiceField.INDICES_PERMISSIONS_KEY)) {
                     // fill in the DLS and FLS permissions for the scroll search action from the scroll context
                     IndicesAccessControl scrollIndicesAccessControl =
-                        readerContext.getFromContext(AuthorizationServiceField.INDICES_PERMISSIONS_KEY);
+                            readerContext.getFromContext(AuthorizationServiceField.INDICES_PERMISSIONS_KEY);
                     assert scrollIndicesAccessControl != null : "scroll does not contain index access control";
                     securityContext.getThreadContext().putTransient(AuthorizationServiceField.INDICES_PERMISSIONS_KEY,
                             scrollIndicesAccessControl);
@@ -93,24 +93,22 @@ public void validateSearchContext(ReaderContext readerContext, TransportRequest
 
     @Override
     public void onPreFetchPhase(SearchContext searchContext) {
-        ensureIndicesAccessControlForScrollThreadContext(searchContext.readerContext());
+        ensureIndicesAccessControlForScrollThreadContext(searchContext);
     }
 
     @Override
     public void onPreQueryPhase(SearchContext searchContext) {
-        ensureIndicesAccessControlForScrollThreadContext(searchContext.readerContext());
+        ensureIndicesAccessControlForScrollThreadContext(searchContext);
     }
 
-    void ensureIndicesAccessControlForScrollThreadContext(ReaderContext readerContext) {
-        if (licenseState.isSecurityEnabled() && readerContext.scrollContext() != null) {
-            IndicesAccessControl scrollIndicesAccessControl =
-                readerContext.getFromContext(AuthorizationServiceField.INDICES_PERMISSIONS_KEY);
+    void ensureIndicesAccessControlForScrollThreadContext(SearchContext searchContext) {
+        if (licenseState.isSecurityEnabled() && searchContext.readerContext().scrollContext() != null) {
             IndicesAccessControl threadIndicesAccessControl =
                     securityContext.getThreadContext().getTransient(AuthorizationServiceField.INDICES_PERMISSIONS_KEY);
-            if (scrollIndicesAccessControl != threadIndicesAccessControl) {
-                throw new ElasticsearchSecurityException("[" + readerContext.id() + "] expected scroll indices access control [" +
-                        scrollIndicesAccessControl.toString() + "] but found [" + threadIndicesAccessControl.toString() + "] in thread " +
-                        "context");
+            if (null == threadIndicesAccessControl) {
+                throw new ElasticsearchSecurityException("Unexpected null indices access control for search context [" +
+                        searchContext.id() + "] for request [" + searchContext.request().getDescription() + "] with source [" +
+                        searchContext.source() + "]");
             }
         }
     }
@@ -131,7 +129,7 @@ static void ensureAuthenticatedUserIsSame(Authentication original, Authenticatio
         if (original.getUser().isRunAs()) {
             if (current.getUser().isRunAs()) {
                 sameRealmType = original.getLookedUpBy().getType().equals(current.getLookedUpBy().getType());
-            }  else {
+            } else {
                 sameRealmType = original.getLookedUpBy().getType().equals(current.getAuthenticatedBy().getType());
             }
         } else if (current.getUser().isRunAs()) {