Avoid loadAuthorizedIndices for requests with concrete names (#81237)

This PR replaces the concrete list of authorizeIndices with a lazily load class
so that loading is not triggered if not necessary.

For example, a search targeting a concret name, GET index/_search no longer
triggers the loading. But names with wildcard will still trigger the loading.

However if we organize the indices with data streams and alias, it is possible
to achieve similar wildcard effect while still avoid the loading. That is,
searches like GET alias/_search or GET data_stream/_search will not
trigger the loading even they target multiple indices behind the single name.

docs/changelog/81237.yaml

@@ -0,0 +1,5 @@
+pr: 81237
+summary: Avoid loading authorized indices when requested indices are all concrete names
+area: Authorization
+type: enhancement
+issues: []

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

@@ -129,7 +129,7 @@ public class AuthorizationService {
     private final Set<RequestInterceptor> requestInterceptors;
     private final XPackLicenseState licenseState;
     private final OperatorPrivilegesService operatorPrivilegesService;
-    private final LoadAuthorizedIndicesTimeChecker.Factory authzIndicesTimerFactory;
+
     private final boolean isAnonymousEnabled;
     private final boolean anonymousAuthzExceptionEnabled;
 
@@ -155,13 +155,16 @@ public AuthorizationService(
         this.anonymousUser = anonymousUser;
         this.isAnonymousEnabled = AnonymousUser.isAnonymousEnabled(settings);
         this.anonymousAuthzExceptionEnabled = ANONYMOUS_AUTHORIZATION_EXCEPTION_SETTING.get(settings);
-        this.rbacEngine = new RBACEngine(settings, rolesStore);
+        this.rbacEngine = new RBACEngine(
+            settings,
+            rolesStore,
+            new LoadAuthorizedIndicesTimeChecker.Factory(logger, settings, clusterService.getClusterSettings())
+        );
         this.authorizationEngine = authorizationEngine == null ? this.rbacEngine : authorizationEngine;
         this.requestInterceptors = requestInterceptors;
         this.settings = settings;
         this.licenseState = licenseState;
         this.operatorPrivilegesService = operatorPrivilegesService;
-        this.authzIndicesTimerFactory = new LoadAuthorizedIndicesTimeChecker.Factory(logger, settings, clusterService.getClusterSettings());
     }
 
     public void checkPrivileges(
@@ -404,24 +407,15 @@ private void authorizeAction(
             }, clusterAuthzListener::onFailure));
         } else if (isIndexAction(action)) {
             final Metadata metadata = clusterService.state().metadata();
-            final AsyncSupplier<Set<String>> authorizedIndicesSupplier = new CachingAsyncSupplier<>(authzIndicesListener -> {
-                Consumer<Collection<String>> timeChecker = authzIndicesTimerFactory.newTimer(requestInfo);
-                authzEngine.loadAuthorizedIndices(
-                    requestInfo,
-                    authzInfo,
-                    metadata.getIndicesLookup(),
-                    authzIndicesListener.map(authzIndices -> {
-                        timeChecker.accept(authzIndices);
-                        return authzIndices;
-                    })
-                );
-            });
             final AsyncSupplier<ResolvedIndices> resolvedIndicesAsyncSupplier = new CachingAsyncSupplier<>(resolvedIndicesListener -> {
                 final ResolvedIndices resolvedIndices = indicesAndAliasesResolver.tryResolveWithoutWildcards(action, request);
                 if (resolvedIndices != null) {
                     resolvedIndicesListener.onResponse(resolvedIndices);
                 } else {
-                    authorizedIndicesSupplier.getAsync(
+                    authzEngine.loadAuthorizedIndices(
+                        requestInfo,
+                        authzInfo,
+                        metadata.getIndicesLookup(),
                         ActionListener.wrap(
                             authorizedIndices -> resolvedIndicesListener.onResponse(
                                 indicesAndAliasesResolver.resolve(action, request, metadata, authorizedIndices)

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

@@ -83,14 +83,17 @@
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.Iterator;
 import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Objects;
 import java.util.Set;
 import java.util.TreeSet;
+import java.util.function.Consumer;
 import java.util.function.Predicate;
+import java.util.function.Supplier;
 
 import static org.elasticsearch.common.Strings.arrayToCommaDelimitedString;
 import static org.elasticsearch.xpack.security.action.user.TransportHasPrivilegesAction.getApplicationNames;
@@ -114,10 +117,16 @@ public class RBACEngine implements AuthorizationEngine {
 
     private final CompositeRolesStore rolesStore;
     private final FieldPermissionsCache fieldPermissionsCache;
+    private final LoadAuthorizedIndicesTimeChecker.Factory authzIndicesTimerFactory;
 
-    public RBACEngine(Settings settings, CompositeRolesStore rolesStore) {
+    public RBACEngine(
+        Settings settings,
+        CompositeRolesStore rolesStore,
+        LoadAuthorizedIndicesTimeChecker.Factory authzIndicesTimerFactory
+    ) {
         this.rolesStore = rolesStore;
         this.fieldPermissionsCache = new FieldPermissionsCache(settings);
+        this.authzIndicesTimerFactory = authzIndicesTimerFactory;
     }
 
     @Override
@@ -458,7 +467,9 @@ public void loadAuthorizedIndices(
     ) {
         if (authorizationInfo instanceof RBACAuthorizationInfo) {
             final Role role = ((RBACAuthorizationInfo) authorizationInfo).getRole();
-            listener.onResponse(resolveAuthorizedIndicesFromRole(role, requestInfo, indicesLookup));
+            listener.onResponse(
+                resolveAuthorizedIndicesFromRole(role, requestInfo, indicesLookup, () -> authzIndicesTimerFactory.newTimer(requestInfo))
+            );
         } else {
             listener.onFailure(
                 new IllegalArgumentException("unsupported authorization info:" + authorizationInfo.getClass().getSimpleName())
@@ -659,34 +670,64 @@ GetUserPrivilegesResponse buildUserPrivilegesResponseObject(Role userRole) {
     }
 
     static Set<String> resolveAuthorizedIndicesFromRole(Role role, RequestInfo requestInfo, Map<String, IndexAbstraction> lookup) {
+        return resolveAuthorizedIndicesFromRole(role, requestInfo, lookup, () -> LoadAuthorizedIndicesTimeChecker.NO_OP_CONSUMER);
+    }
+
+    static Set<String> resolveAuthorizedIndicesFromRole(
+        Role role,
+        RequestInfo requestInfo,
+        Map<String, IndexAbstraction> lookup,
+        Supplier<Consumer<Collection<String>>> timerSupplier
+    ) {
         Predicate<IndexAbstraction> predicate = role.allowedIndicesMatcher(requestInfo.getAction());
 
         // do not include data streams for actions that do not operate on data streams
         TransportRequest request = requestInfo.getRequest();
         final boolean includeDataStreams = (request instanceof IndicesRequest) && ((IndicesRequest) request).includeDataStreams();
 
-        Set<String> indicesAndAliases = new HashSet<>();
-        // TODO: can this be done smarter? I think there are usually more indices/aliases in the cluster then indices defined a roles?
-        if (includeDataStreams) {
-            for (IndexAbstraction indexAbstraction : lookup.values()) {
-                if (predicate.test(indexAbstraction)) {
-                    indicesAndAliases.add(indexAbstraction.getName());
-                    if (indexAbstraction.getType() == IndexAbstraction.Type.DATA_STREAM) {
-                        // add data stream and its backing indices for any authorized data streams
-                        for (Index index : indexAbstraction.getIndices()) {
-                            indicesAndAliases.add(index.getName());
+        return new AuthorizedIndicesSet(() -> {
+            Consumer<Collection<String>> timeChecker = timerSupplier.get();
+            Set<String> indicesAndAliases = new HashSet<>();
+            // TODO: can this be done smarter? I think there are usually more indices/aliases in the cluster then indices defined a roles?
+            if (includeDataStreams) {
+                for (IndexAbstraction indexAbstraction : lookup.values()) {
+                    if (predicate.test(indexAbstraction)) {
+                        indicesAndAliases.add(indexAbstraction.getName());
+                        if (indexAbstraction.getType() == IndexAbstraction.Type.DATA_STREAM) {
+                            // add data stream and its backing indices for any authorized data streams
+                            for (Index index : indexAbstraction.getIndices()) {
+                                indicesAndAliases.add(index.getName());
+                            }
                         }
                     }
                 }
+            } else {
+                for (IndexAbstraction indexAbstraction : lookup.values()) {
+                    if (indexAbstraction.getType() != IndexAbstraction.Type.DATA_STREAM && predicate.test(indexAbstraction)) {
+                        indicesAndAliases.add(indexAbstraction.getName());
+                    }
+                }
             }
-        } else {
-            for (IndexAbstraction indexAbstraction : lookup.values()) {
-                if (indexAbstraction.getType() != IndexAbstraction.Type.DATA_STREAM && predicate.test(indexAbstraction)) {
-                    indicesAndAliases.add(indexAbstraction.getName());
+            timeChecker.accept(indicesAndAliases);
+            return indicesAndAliases;
+        }, name -> {
+            final IndexAbstraction indexAbstraction = lookup.get(name);
+            if (indexAbstraction == null) {
+                return false;
+            }
+            if (includeDataStreams) {
+                // We check the parent data stream first if there is one. For testing requested indices, this is most likely
+                // more efficient than checking the index name first because we recommend grant privileges over data stream
+                // instead of backing indices.
+                if (indexAbstraction.getParentDataStream() != null && predicate.test(indexAbstraction.getParentDataStream())) {
+                    return true;
+                } else {
+                    return predicate.test(indexAbstraction);
                 }
+            } else {
+                return indexAbstraction.getType() != IndexAbstraction.Type.DATA_STREAM && predicate.test(indexAbstraction);
             }
-        }
-        return Collections.unmodifiableSet(indicesAndAliases);
+        });
     }
 
     private IndexAuthorizationResult buildIndicesAccessControl(
@@ -813,4 +854,104 @@ private static boolean isAsyncRelatedAction(String action) {
             || action.equals(EqlAsyncActionNames.EQL_ASYNC_GET_RESULT_ACTION_NAME)
             || action.equals(SqlAsyncActionNames.SQL_ASYNC_GET_RESULT_ACTION_NAME);
     }
+
+    /**
+     * A lazily loaded Set for authorized indices. It avoids loading the set if only contains check is required.
+     * It only loads the set if iterating through it is necessary, i.e. when expanding wildcards.
+     *
+     * NOTE that the lazy loading is NOT thread-safe and must NOT be used by multi-threads.
+     * The current usage has it wrapped inside a CachingAsyncSupplier which guarantees it to be accessed
+     * from a single thread. Extra caution is needed if moving or using this class in other places.
+     */
+    static class AuthorizedIndicesSet implements Set<String> {
+
+        private final Supplier<Set<String>> supplier;
+        private final Predicate<String> predicate;
+        private Set<String> authorizedIndices = null;
+
+        AuthorizedIndicesSet(Supplier<Set<String>> supplier, Predicate<String> predicate) {
+            this.supplier = Objects.requireNonNull(supplier);
+            this.predicate = Objects.requireNonNull(predicate);
+        }
+
+        private Set<String> getAuthorizedIndices() {
+            if (authorizedIndices == null) {
+                authorizedIndices = supplier.get();
+            }
+            return authorizedIndices;
+        }
+
+        @Override
+        public int size() {
+            return getAuthorizedIndices().size();
+        }
+
+        @Override
+        public boolean isEmpty() {
+            return getAuthorizedIndices().isEmpty();
+        }
+
+        @Override
+        public boolean contains(Object o) {
+            if (authorizedIndices == null) {
+                return predicate.test((String) o);
+            } else {
+                return authorizedIndices.contains(o);
+            }
+        }
+
+        @Override
+        public Iterator<String> iterator() {
+            return getAuthorizedIndices().iterator();
+        }
+
+        @Override
+        public Object[] toArray() {
+            return getAuthorizedIndices().toArray();
+        }
+
+        @Override
+        public <T> T[] toArray(T[] a) {
+            return getAuthorizedIndices().toArray(a);
+        }
+
+        @Override
+        public boolean add(String s) {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public boolean remove(Object o) {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public boolean containsAll(Collection<?> c) {
+            if (authorizedIndices == null) {
+                return c.stream().allMatch(this::contains);
+            } else {
+                return authorizedIndices.containsAll(c);
+            }
+        }
+
+        @Override
+        public boolean addAll(Collection<? extends String> c) {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public boolean retainAll(Collection<?> c) {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public boolean removeAll(Collection<?> c) {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public void clear() {
+            throw new UnsupportedOperationException();
+        }
+    }
 }