From 50ed781c9f0266d473774c4fdac3546b79cab87a Mon Sep 17 00:00:00 2001
From: Dan Hermann <danhermann@users.noreply.github.com>
Date: Thu, 2 Jul 2020 21:56:16 -0500
Subject: [PATCH] Mirror privileges over data streams to their backing indices
 (#58381)

---
 .../runConfigurations/Debug_Elasticsearch.xml |   4 +-
 .../20_unsupported_apis.yml                   |   9 -
 .../elasticsearch/action/IndicesRequest.java  |   9 +
 .../shards/ClusterSearchShardsRequest.java    |   5 +
 .../indices/stats/IndicesStatsRequest.java    |   5 +
 .../fieldcaps/FieldCapabilitiesRequest.java   |   5 +
 .../action/search/SearchRequest.java          |   5 +
 .../authz/permission/IndicesPermission.java   |  11 +-
 .../authz/IndicesAndAliasesResolver.java      |  28 +-
 .../xpack/security/authz/RBACEngine.java      |  30 +-
 .../authz/AuthorizedIndicesTests.java         |  28 +-
 .../authz/IndicesAndAliasesResolverTests.java | 303 +++++++++++++++++-
 .../xpack/security/authz/RBACEngineTests.java |  43 +++
 .../accesscontrol/IndicesPermissionTests.java |  48 +++
 .../test/security/authz/50_data_streams.yml   | 149 +++++++++
 15 files changed, 631 insertions(+), 51 deletions(-)
 create mode 100644 x-pack/plugin/src/test/resources/rest-api-spec/test/security/authz/50_data_streams.yml

diff --git a/.idea/runConfigurations/Debug_Elasticsearch.xml b/.idea/runConfigurations/Debug_Elasticsearch.xml
index 6b58e18d22266..0873e20eaaeb9 100644
--- a/.idea/runConfigurations/Debug_Elasticsearch.xml
+++ b/.idea/runConfigurations/Debug_Elasticsearch.xml
@@ -1,11 +1,11 @@
 <component name="ProjectRunConfigurationManager">
   <configuration default="false" name="Debug Elasticsearch" type="Remote">
     <option name="USE_SOCKET_TRANSPORT" value="true" />
-    <option name="SERVER_MODE" value="true" />
+    <option name="SERVER_MODE" value="false" />
     <option name="SHMEM_ADDRESS" />
     <option name="HOST" value="localhost" />
     <option name="PORT" value="5005" />
-    <option name="AUTO_RESTART" value="true" />
+    <option name="AUTO_RESTART" value="false" />
     <RunnerSettings RunnerId="Debug">
       <option name="DEBUG_PORT" value="5005" />
       <option name="LOCAL" value="false" />
diff --git a/rest-api-spec/src/main/resources/rest-api-spec/test/indices.data_stream/20_unsupported_apis.yml b/rest-api-spec/src/main/resources/rest-api-spec/test/indices.data_stream/20_unsupported_apis.yml
index 256b04ab944e5..84c225931e504 100644
--- a/rest-api-spec/src/main/resources/rest-api-spec/test/indices.data_stream/20_unsupported_apis.yml
+++ b/rest-api-spec/src/main/resources/rest-api-spec/test/indices.data_stream/20_unsupported_apis.yml
@@ -47,15 +47,6 @@
       indices.delete:
         index: logs-foobar
 
-  - do:
-      indices.create:
-        index: logs-foobarbaz
-
-  - do:
-      catch: bad_request
-      indices.close:
-        index: logs-*
-
   - do:
       indices.delete_data_stream:
         name: logs-foobar
diff --git a/server/src/main/java/org/elasticsearch/action/IndicesRequest.java b/server/src/main/java/org/elasticsearch/action/IndicesRequest.java
index 3ef699818b6bf..bd6421ad0fc0b 100644
--- a/server/src/main/java/org/elasticsearch/action/IndicesRequest.java
+++ b/server/src/main/java/org/elasticsearch/action/IndicesRequest.java
@@ -40,6 +40,15 @@ public interface IndicesRequest {
      */
     IndicesOptions indicesOptions();
 
+    /**
+     * Determines whether the request should be applied to data streams. When {@code false}, none of the names or
+     * wildcard expressions in {@link #indices} should be applied to or expanded to any data streams. All layers
+     * involved in the request's fulfillment including security, name resolution, etc., should respect this flag.
+     */
+    default boolean includeDataStreams() {
+        return false;
+    }
+
     interface Replaceable extends IndicesRequest {
         /**
          * Sets the indices that the action relates to.
diff --git a/server/src/main/java/org/elasticsearch/action/admin/cluster/shards/ClusterSearchShardsRequest.java b/server/src/main/java/org/elasticsearch/action/admin/cluster/shards/ClusterSearchShardsRequest.java
index e1ba4b76e4168..d9ab03c9440eb 100644
--- a/server/src/main/java/org/elasticsearch/action/admin/cluster/shards/ClusterSearchShardsRequest.java
+++ b/server/src/main/java/org/elasticsearch/action/admin/cluster/shards/ClusterSearchShardsRequest.java
@@ -112,6 +112,11 @@ public ClusterSearchShardsRequest indicesOptions(IndicesOptions indicesOptions)
         return this;
     }
 
+    @Override
+    public boolean includeDataStreams() {
+        return true;
+    }
+
     /**
      * A comma separated list of routing values to control the shards the search will be executed on.
      */
diff --git a/server/src/main/java/org/elasticsearch/action/admin/indices/stats/IndicesStatsRequest.java b/server/src/main/java/org/elasticsearch/action/admin/indices/stats/IndicesStatsRequest.java
index 6af50d5e2eacb..73d05a18ddb04 100644
--- a/server/src/main/java/org/elasticsearch/action/admin/indices/stats/IndicesStatsRequest.java
+++ b/server/src/main/java/org/elasticsearch/action/admin/indices/stats/IndicesStatsRequest.java
@@ -281,4 +281,9 @@ public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
         flags.writeTo(out);
     }
+
+    @Override
+    public boolean includeDataStreams() {
+        return true;
+    }
 }
diff --git a/server/src/main/java/org/elasticsearch/action/fieldcaps/FieldCapabilitiesRequest.java b/server/src/main/java/org/elasticsearch/action/fieldcaps/FieldCapabilitiesRequest.java
index 520a53dcdea6f..62629bfbbfc3f 100644
--- a/server/src/main/java/org/elasticsearch/action/fieldcaps/FieldCapabilitiesRequest.java
+++ b/server/src/main/java/org/elasticsearch/action/fieldcaps/FieldCapabilitiesRequest.java
@@ -157,6 +157,11 @@ public IndicesOptions indicesOptions() {
         return indicesOptions;
     }
 
+    @Override
+    public boolean includeDataStreams() {
+        return true;
+    }
+
     public boolean includeUnmapped() {
         return includeUnmapped;
     }
diff --git a/server/src/main/java/org/elasticsearch/action/search/SearchRequest.java b/server/src/main/java/org/elasticsearch/action/search/SearchRequest.java
index 5dd8438b01eff..15168091289c7 100644
--- a/server/src/main/java/org/elasticsearch/action/search/SearchRequest.java
+++ b/server/src/main/java/org/elasticsearch/action/search/SearchRequest.java
@@ -348,6 +348,11 @@ public SearchRequest indicesOptions(IndicesOptions indicesOptions) {
         return this;
     }
 
+    @Override
+    public boolean includeDataStreams() {
+        return true;
+    }
+
     /**
      * Returns whether network round-trips should be minimized when executing cross-cluster search requests.
      * Defaults to <code>true</code>.
diff --git a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authz/permission/IndicesPermission.java b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authz/permission/IndicesPermission.java
index 0e8e1e75b0e72..bbc664f50f8c6 100644
--- a/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authz/permission/IndicesPermission.java
+++ b/x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authz/permission/IndicesPermission.java
@@ -194,7 +194,7 @@ public Automaton allowedActionsMatcher(String index) {
      * Authorizes the provided action against the provided indices, given the current cluster metadata
      */
     public Map<String, IndicesAccessControl.IndexAccessControl> authorize(String action, Set<String> requestedIndicesOrAliases,
-                                                                          Map<String, IndexAbstraction> allAliasesAndIndices,
+                                                                          Map<String, IndexAbstraction> lookup,
                                                                           FieldPermissionsCache fieldPermissionsCache) {
         // now... every index that is associated with the request, must be granted
         // by at least one indices permission group
@@ -205,7 +205,7 @@ public Map<String, IndicesAccessControl.IndexAccessControl> authorize(String act
         for (String indexOrAlias : requestedIndicesOrAliases) {
             boolean granted = false;
             Set<String> concreteIndices = new HashSet<>();
-            IndexAbstraction indexAbstraction = allAliasesAndIndices.get(indexOrAlias);
+            IndexAbstraction indexAbstraction = lookup.get(indexOrAlias);
             if (indexAbstraction != null) {
                 for (IndexMetadata indexMetadata : indexAbstraction.getIndices()) {
                     concreteIndices.add(indexMetadata.getIndex().getName());
@@ -213,7 +213,12 @@ public Map<String, IndicesAccessControl.IndexAccessControl> authorize(String act
             }
 
             for (Group group : groups) {
-                if (group.check(action, indexOrAlias)) {
+                // check for privilege granted directly on the requested index/alias
+                if (group.check(action, indexOrAlias) ||
+                    // check for privilege granted on parent data stream if a backing index
+                    (indexAbstraction != null && indexAbstraction.getType() == IndexAbstraction.Type.CONCRETE_INDEX &&
+                        indexAbstraction.getParentDataStream() != null &&
+                        group.check(action, indexAbstraction.getParentDataStream().getName()))) {
                     granted = true;
                     for (String index : concreteIndices) {
                         Set<FieldPermissions> fieldPermissions = fieldPermissionsByIndex.computeIfAbsent(index, (k) -> new HashSet<>());
diff --git a/x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/authz/IndicesAndAliasesResolver.java b/x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/authz/IndicesAndAliasesResolver.java
index 31f4e8fe85233..5b0042ab6a422 100644
--- a/x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/authz/IndicesAndAliasesResolver.java
+++ b/x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/authz/IndicesAndAliasesResolver.java
@@ -130,7 +130,7 @@ ResolvedIndices resolveIndicesAndAliases(IndicesRequest indicesRequest, Metadata
             if (IndexNameExpressionResolver.isAllIndices(indicesList(indicesRequest.indices()))) {
                 if (replaceWildcards) {
                     for (String authorizedIndex : authorizedIndices) {
-                        if (isIndexVisible("*", authorizedIndex, indicesOptions, metadata)) {
+                        if (isIndexVisible("*", authorizedIndex, indicesOptions, metadata, indicesRequest.includeDataStreams())) {
                             resolvedIndicesBuilder.addLocal(authorizedIndex);
                         }
                     }
@@ -145,7 +145,7 @@ ResolvedIndices resolveIndicesAndAliases(IndicesRequest indicesRequest, Metadata
                     split = new ResolvedIndices(Arrays.asList(indicesRequest.indices()), Collections.emptyList());
                 }
                 List<String> replaced = replaceWildcardsWithAuthorizedIndices(split.getLocal(), indicesOptions, metadata,
-                        authorizedIndices, replaceWildcards);
+                        authorizedIndices, replaceWildcards, indicesRequest.includeDataStreams());
                 if (indicesOptions.ignoreUnavailable()) {
                     //out of all the explicit names (expanded from wildcards and original ones that were left untouched)
                     //remove all the ones that the current user is not authorized for and ignore them
@@ -337,7 +337,8 @@ private boolean containsWildcards(IndicesRequest indicesRequest) {
 
     //TODO Investigate reusing code from vanilla es to resolve index names and wildcards
     private List<String> replaceWildcardsWithAuthorizedIndices(Iterable<String> indices, IndicesOptions indicesOptions, Metadata metadata,
-                                                               List<String> authorizedIndices, boolean replaceWildcards) {
+                                                               List<String> authorizedIndices, boolean replaceWildcards,
+                                                               boolean includeDataStreams) {
         //the order matters when it comes to exclusions
         List<String> finalIndices = new ArrayList<>();
         boolean wildcardSeen = false;
@@ -359,7 +360,7 @@ private List<String> replaceWildcardsWithAuthorizedIndices(Iterable<String> indi
                     // continue
                     aliasOrIndex = dateMathName;
                 } else if (authorizedIndices.contains(dateMathName) &&
-                    isIndexVisible(aliasOrIndex, dateMathName, indicesOptions, metadata, true)) {
+                    isIndexVisible(aliasOrIndex, dateMathName, indicesOptions, metadata, includeDataStreams, true)) {
                     if (minus) {
                         finalIndices.remove(dateMathName);
                     } else {
@@ -377,7 +378,7 @@ private List<String> replaceWildcardsWithAuthorizedIndices(Iterable<String> indi
                 Set<String> resolvedIndices = new HashSet<>();
                 for (String authorizedIndex : authorizedIndices) {
                     if (Regex.simpleMatch(aliasOrIndex, authorizedIndex) &&
-                        isIndexVisible(aliasOrIndex, authorizedIndex, indicesOptions, metadata)) {
+                        isIndexVisible(aliasOrIndex, authorizedIndex, indicesOptions, metadata, includeDataStreams)) {
                         resolvedIndices.add(authorizedIndex);
                     }
                 }
@@ -412,13 +413,17 @@ private List<String> replaceWildcardsWithAuthorizedIndices(Iterable<String> indi
         return finalIndices;
     }
 
-    private static boolean isIndexVisible(String expression, String index, IndicesOptions indicesOptions, Metadata metadata) {
-        return isIndexVisible(expression, index, indicesOptions, metadata, false);
+    private static boolean isIndexVisible(String expression, String index, IndicesOptions indicesOptions, Metadata metadata,
+                                          boolean includeDataStreams) {
+        return isIndexVisible(expression, index, indicesOptions, metadata, includeDataStreams, false);
     }
 
     private static boolean isIndexVisible(String expression, String index, IndicesOptions indicesOptions, Metadata metadata,
-                                          boolean dateMathExpression) {
+                                          boolean includeDataStreams, boolean dateMathExpression) {
         IndexAbstraction indexAbstraction = metadata.getIndicesLookup().get(index);
+        if (indexAbstraction == null) {
+            throw new IllegalStateException("could not resolve index abstraction [" + index + "]");
+        }
         final boolean isHidden = indexAbstraction.isHidden();
         if (indexAbstraction.getType() == IndexAbstraction.Type.ALIAS) {
             //it's an alias, ignore expandWildcardsOpen and expandWildcardsClosed.
@@ -433,12 +438,7 @@ private static boolean isIndexVisible(String expression, String index, IndicesOp
             }
         }
         if (indexAbstraction.getType() == IndexAbstraction.Type.DATA_STREAM) {
-            // If indicesOptions.includeDataStreams() returns false then we fail later in IndexNameExpressionResolver.
-            if (isHidden == false || indicesOptions.expandWildcardsHidden()) {
-                return true;
-            } else {
-                return false;
-            }
+            return includeDataStreams;
         }
         assert indexAbstraction.getIndices().size() == 1 : "concrete index must point to a single index";
         IndexMetadata indexMetadata = indexAbstraction.getIndices().get(0);
diff --git a/x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/authz/RBACEngine.java b/x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/authz/RBACEngine.java
index 8098235c0ff33..99764879b80ba 100644
--- a/x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/authz/RBACEngine.java
+++ b/x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/authz/RBACEngine.java
@@ -86,6 +86,7 @@
 import java.util.Set;
 import java.util.TreeSet;
 import java.util.function.Predicate;
+import java.util.stream.Collectors;
 
 import static org.elasticsearch.common.Strings.arrayToCommaDelimitedString;
 import static org.elasticsearch.xpack.security.action.user.TransportHasPrivilegesAction.getApplicationNames;
@@ -343,7 +344,7 @@ public void loadAuthorizedIndices(RequestInfo requestInfo, AuthorizationInfo aut
                                       Map<String, IndexAbstraction> indicesLookup, ActionListener<List<String>> listener) {
         if (authorizationInfo instanceof RBACAuthorizationInfo) {
             final Role role = ((RBACAuthorizationInfo) authorizationInfo).getRole();
-            listener.onResponse(resolveAuthorizedIndicesFromRole(role, requestInfo.getAction(), indicesLookup));
+            listener.onResponse(resolveAuthorizedIndicesFromRole(role, requestInfo, indicesLookup));
         } else {
             listener.onFailure(
                 new IllegalArgumentException("unsupported authorization info:" + authorizationInfo.getClass().getSimpleName()));
@@ -500,18 +501,29 @@ GetUserPrivilegesResponse buildUserPrivilegesResponseObject(Role userRole) {
         return new GetUserPrivilegesResponse(cluster, conditionalCluster, indices, application, runAs);
     }
 
-    static List<String> resolveAuthorizedIndicesFromRole(Role role, String action, Map<String, IndexAbstraction> aliasAndIndexLookup) {
-        Predicate<String> predicate = role.allowedIndicesMatcher(action);
+    static List<String> resolveAuthorizedIndicesFromRole(Role role, RequestInfo requestInfo, Map<String, IndexAbstraction> lookup) {
+        Predicate<String> predicate = role.allowedIndicesMatcher(requestInfo.getAction());
 
-        List<String> indicesAndAliases = new ArrayList<>();
+        // do not include data streams for actions that do not operate on data streams
+        TransportRequest request = requestInfo.getRequest();
+        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?
-        for (Map.Entry<String, IndexAbstraction> entry : aliasAndIndexLookup.entrySet()) {
-            String aliasOrIndex = entry.getKey();
-            if (predicate.test(aliasOrIndex)) {
-                indicesAndAliases.add(aliasOrIndex);
+        for (Map.Entry<String, IndexAbstraction> entry : lookup.entrySet()) {
+            String indexAbstraction = entry.getKey();
+            if (predicate.test(indexAbstraction)) {
+                if (entry.getValue().getType() != IndexAbstraction.Type.DATA_STREAM) {
+                    indicesAndAliases.add(indexAbstraction);
+                } else if (includeDataStreams) {
+                    // add data stream and its backing indices for any authorized data streams
+                    indicesAndAliases.addAll(entry.getValue().getIndices().stream()
+                        .map(i -> i.getIndex().getName()).collect(Collectors.toList()));
+                    indicesAndAliases.add(indexAbstraction);
+                }
             }
         }
-        return Collections.unmodifiableList(indicesAndAliases);
+        return Collections.unmodifiableList(new ArrayList<>(indicesAndAliases));
     }
 
     private void buildIndicesAccessControl(Authentication authentication, String action,
diff --git a/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authz/AuthorizedIndicesTests.java b/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authz/AuthorizedIndicesTests.java
index c2fa04af08002..e7fca9b27d1ad 100644
--- a/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authz/AuthorizedIndicesTests.java
+++ b/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authz/AuthorizedIndicesTests.java
@@ -14,6 +14,8 @@
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.set.Sets;
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.transport.TransportRequest;
+import org.elasticsearch.xpack.core.security.authz.AuthorizationEngine;
 import org.elasticsearch.xpack.core.security.authz.RoleDescriptor;
 import org.elasticsearch.xpack.core.security.authz.RoleDescriptor.IndicesPrivileges;
 import org.elasticsearch.xpack.core.security.authz.permission.FieldPermissions;
@@ -35,7 +37,7 @@ public class AuthorizedIndicesTests extends ESTestCase {
 
     public void testAuthorizedIndicesUserWithoutRoles() {
         List<String> authorizedIndices =
-            RBACEngine.resolveAuthorizedIndicesFromRole(Role.EMPTY, "", Metadata.EMPTY_METADATA.getIndicesLookup());
+            RBACEngine.resolveAuthorizedIndicesFromRole(Role.EMPTY, getRequestInfo(""), Metadata.EMPTY_METADATA.getIndicesLookup());
         assertTrue(authorizedIndices.isEmpty());
     }
 
@@ -71,7 +73,7 @@ public void testAuthorizedIndicesUserWithSomeRoles() {
         CompositeRolesStore.buildRoleFromDescriptors(descriptors, new FieldPermissionsCache(Settings.EMPTY), null, future);
         Role roles = future.actionGet();
         List<String> list =
-            RBACEngine.resolveAuthorizedIndicesFromRole(roles, SearchAction.NAME, metadata.getIndicesLookup());
+            RBACEngine.resolveAuthorizedIndicesFromRole(roles, getRequestInfo(SearchAction.NAME), metadata.getIndicesLookup());
         assertThat(list, containsInAnyOrder("a1", "a2", "aaaaaa", "b", "ab"));
         assertFalse(list.contains("bbbbb"));
         assertFalse(list.contains("ba"));
@@ -81,15 +83,15 @@ public void testAuthorizedIndicesUserWithSomeRoles() {
 
     public void testAuthorizedIndicesUserWithSomeRolesEmptyMetadata() {
         Role role = Role.builder("role").add(IndexPrivilege.ALL, "*").build();
-        List<String> authorizedIndices =
-            RBACEngine.resolveAuthorizedIndicesFromRole(role, SearchAction.NAME, Metadata.EMPTY_METADATA.getIndicesLookup());
+        List<String> authorizedIndices = RBACEngine.resolveAuthorizedIndicesFromRole(role, getRequestInfo(SearchAction.NAME),
+            Metadata.EMPTY_METADATA.getIndicesLookup());
         assertTrue(authorizedIndices.isEmpty());
     }
 
     public void testSecurityIndicesAreRemovedFromRegularUser() {
         Role role = Role.builder("user_role").add(IndexPrivilege.ALL, "*").cluster(Set.of("all"), Set.of()).build();
-        List<String> authorizedIndices =
-            RBACEngine.resolveAuthorizedIndicesFromRole(role, SearchAction.NAME, Metadata.EMPTY_METADATA.getIndicesLookup());
+        List<String> authorizedIndices = RBACEngine.resolveAuthorizedIndicesFromRole(role, getRequestInfo(SearchAction.NAME),
+            Metadata.EMPTY_METADATA.getIndicesLookup());
         assertTrue(authorizedIndices.isEmpty());
     }
 
@@ -114,7 +116,7 @@ public void testSecurityIndicesAreRestrictedForDefaultRole() {
                 .build();
 
         List<String> authorizedIndices =
-            RBACEngine.resolveAuthorizedIndicesFromRole(role, SearchAction.NAME, metadata.getIndicesLookup());
+            RBACEngine.resolveAuthorizedIndicesFromRole(role, getRequestInfo(SearchAction.NAME), metadata.getIndicesLookup());
         assertThat(authorizedIndices, containsInAnyOrder("an-index", "another-index"));
         assertThat(authorizedIndices, not(contains(internalSecurityIndex)));
         assertThat(authorizedIndices, not(contains(RestrictedIndicesNames.SECURITY_MAIN_ALIAS)));
@@ -140,13 +142,21 @@ public void testSecurityIndicesAreNotRemovedFromUnrestrictedRole() {
                 .build();
 
         List<String> authorizedIndices =
-            RBACEngine.resolveAuthorizedIndicesFromRole(role, SearchAction.NAME, metadata.getIndicesLookup());
+            RBACEngine.resolveAuthorizedIndicesFromRole(role, getRequestInfo(SearchAction.NAME), metadata.getIndicesLookup());
         assertThat(authorizedIndices, containsInAnyOrder(
             "an-index", "another-index", RestrictedIndicesNames.SECURITY_MAIN_ALIAS, internalSecurityIndex));
 
         List<String> authorizedIndicesSuperUser =
-            RBACEngine.resolveAuthorizedIndicesFromRole(role, SearchAction.NAME, metadata.getIndicesLookup());
+            RBACEngine.resolveAuthorizedIndicesFromRole(role, getRequestInfo(SearchAction.NAME), metadata.getIndicesLookup());
         assertThat(authorizedIndicesSuperUser, containsInAnyOrder(
             "an-index", "another-index", RestrictedIndicesNames.SECURITY_MAIN_ALIAS, internalSecurityIndex));
     }
+
+    public static AuthorizationEngine.RequestInfo getRequestInfo(String action) {
+        return getRequestInfo(TransportRequest.Empty.INSTANCE, action);
+    }
+
+    public static AuthorizationEngine.RequestInfo getRequestInfo(TransportRequest request, String action) {
+        return new AuthorizationEngine.RequestInfo(null, request, action);
+    }
 }
diff --git a/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authz/IndicesAndAliasesResolverTests.java b/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authz/IndicesAndAliasesResolverTests.java
index fcb94d8b53380..16e5505a286e8 100644
--- a/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authz/IndicesAndAliasesResolverTests.java
+++ b/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authz/IndicesAndAliasesResolverTests.java
@@ -81,6 +81,9 @@
 
 import static org.elasticsearch.cluster.DataStreamTestHelper.createTimestampField;
 import static org.elasticsearch.xpack.core.security.index.RestrictedIndicesNames.SECURITY_MAIN_ALIAS;
+import static org.elasticsearch.xpack.security.authz.AuthorizedIndicesTests.getRequestInfo;
+import static org.hamcrest.CoreMatchers.instanceOf;
+import static org.hamcrest.CoreMatchers.is;
 import static org.hamcrest.Matchers.arrayContaining;
 import static org.hamcrest.Matchers.arrayContainingInAnyOrder;
 import static org.hamcrest.Matchers.contains;
@@ -215,6 +218,27 @@ public void setup() {
                     .privileges("all")
                     .build()
             }, null));
+        roleMap.put("data_stream_test3", new RoleDescriptor("data_stream_test3", null,
+            new IndicesPrivileges[] {
+                IndicesPrivileges.builder()
+                    .indices("logs*")
+                    .privileges("all")
+                    .build()
+            }, null));
+        roleMap.put("backing_index_test_wildcards", new RoleDescriptor("backing_index_test_wildcards", null,
+            new IndicesPrivileges[] {
+                IndicesPrivileges.builder()
+                    .indices(".ds-logs*")
+                    .privileges("all")
+                    .build()
+            }, null));
+        roleMap.put("backing_index_test_name", new RoleDescriptor("backing_index_test_name", null,
+            new IndicesPrivileges[] {
+                IndicesPrivileges.builder()
+                    .indices(dataStreamIndex1.getIndex().getName())
+                    .privileges("all")
+                    .build()
+            }, null));
         final FieldPermissionsCache fieldPermissionsCache = new FieldPermissionsCache(Settings.EMPTY);
         doAnswer((i) -> {
             ActionListener callback =
@@ -1530,13 +1554,13 @@ public void testHiddenAliasesResolution() {
 
     public void testDataStreamResolution() {
         {
-            final User user = new User("data-steam-tester1", "data_stream_test1");
-            final List<String> authorizedIndices = buildAuthorizedIndices(user, SearchAction.NAME);
+            final User user = new User("data-stream-tester1", "data_stream_test1");
 
             // Resolve data streams:
             SearchRequest searchRequest = new SearchRequest();
             searchRequest.indices("logs-*");
             searchRequest.indicesOptions(IndicesOptions.fromOptions(false, false, true, false, false, true, true, true, true));
+            final List<String> authorizedIndices = buildAuthorizedIndices(user, SearchAction.NAME, searchRequest);
             ResolvedIndices resolvedIndices = defaultIndicesResolver.resolveIndicesAndAliases(searchRequest, metadata, authorizedIndices);
             assertThat(resolvedIndices.getLocal(), contains("logs-foobar"));
             assertThat(resolvedIndices.getRemote(), emptyIterable());
@@ -1551,25 +1575,294 @@ public void testDataStreamResolution() {
             assertThat(resolvedIndices.getRemote(), emptyIterable());
         }
         {
-            final User user = new User("data-steam-tester2", "data_stream_test2");
-            final List<String> authorizedIndices = buildAuthorizedIndices(user, SearchAction.NAME);
+            final User user = new User("data-stream-tester2", "data_stream_test2");
 
             // Resolve *all* data streams:
             SearchRequest searchRequest = new SearchRequest();
             searchRequest.indices("logs-*");
             searchRequest.indicesOptions(IndicesOptions.fromOptions(false, false, true, false, false, true, true, true, true));
+            final List<String> authorizedIndices = buildAuthorizedIndices(user, SearchAction.NAME, searchRequest);
             ResolvedIndices resolvedIndices = defaultIndicesResolver.resolveIndicesAndAliases(searchRequest, metadata, authorizedIndices);
             assertThat(resolvedIndices.getLocal(), containsInAnyOrder("logs-foo", "logs-foobar"));
             assertThat(resolvedIndices.getRemote(), emptyIterable());
         }
     }
 
+    public void testDataStreamsAreNotVisibleWhenNotIncludedByRequestWithWildcard() {
+        final User user = new User("data-stream-tester2", "data_stream_test2");
+        GetAliasesRequest request = new GetAliasesRequest("*");
+        assertThat(request, instanceOf(IndicesRequest.Replaceable.class));
+        assertThat(request.includeDataStreams(), is(false));
+
+        // data streams and their backing indices should _not_ be in the authorized list since the backing indices
+        // do not match the requested pattern
+        List<String> dataStreams = List.of("logs-foo", "logs-foobar");
+        final List<String> authorizedIndices = buildAuthorizedIndices(user, GetAliasesAction.NAME, request);
+        for (String dsName : dataStreams) {
+            assertThat(authorizedIndices, not(hasItem(dsName)));
+            DataStream dataStream = metadata.dataStreams().get(dsName);
+            assertThat(authorizedIndices, not(hasItem(dsName)));
+            for (Index i : dataStream.getIndices()) {
+                assertThat(authorizedIndices, not(hasItem(i.getName())));
+            }
+        }
+
+        // neither data streams nor their backing indices will be in the resolved list unless the backing indices matched the requested
+        // pattern
+        ResolvedIndices resolvedIndices = defaultIndicesResolver.resolveIndicesAndAliases(request, metadata, authorizedIndices);
+        for (String dsName : dataStreams) {
+            assertThat(resolvedIndices.getLocal(), not(hasItem(dsName)));
+            DataStream dataStream = metadata.dataStreams().get(dsName);
+            assertThat(resolvedIndices.getLocal(), not(hasItem(dsName)));
+            for (Index i : dataStream.getIndices()) {
+                assertThat(resolvedIndices.getLocal(), not(hasItem(i.getName())));
+            }
+        }
+    }
+
+    public void testDataStreamsAreNotVisibleWhenNotIncludedByRequestWithoutWildcard() {
+        final User user = new User("data-stream-tester2", "data_stream_test2");
+        String dataStreamName = "logs-foobar";
+        GetAliasesRequest request = new GetAliasesRequest(dataStreamName);
+        assertThat(request, instanceOf(IndicesRequest.Replaceable.class));
+        assertThat(request.includeDataStreams(), is(false));
+
+        // data streams and their backing indices should _not_ be in the authorized list since the backing indices
+        // do not match the requested name
+        final List<String> authorizedIndices = buildAuthorizedIndices(user, GetAliasesAction.NAME, request);
+        assertThat(authorizedIndices, not(hasItem(dataStreamName)));
+        DataStream dataStream = metadata.dataStreams().get(dataStreamName);
+        assertThat(authorizedIndices, not(hasItem(dataStreamName)));
+        for (Index i : dataStream.getIndices()) {
+            assertThat(authorizedIndices, not(hasItem(i.getName())));
+        }
+
+        // neither data streams nor their backing indices will be in the resolved list since the backing indices do not match the
+        // requested name(s)
+        ResolvedIndices resolvedIndices = defaultIndicesResolver.resolveIndicesAndAliases(request, metadata, authorizedIndices);
+        assertThat(resolvedIndices.getLocal(), not(hasItem(dataStreamName)));
+        for (Index i : dataStream.getIndices()) {
+            assertThat(resolvedIndices.getLocal(), not(hasItem(i.getName())));
+        }
+    }
+
+    public void testDataStreamsAreVisibleWhenIncludedByRequestWithWildcard() {
+        final User user = new User("data-stream-tester3", "data_stream_test3");
+        SearchRequest request = new SearchRequest("logs*");
+        assertThat(request, instanceOf(IndicesRequest.Replaceable.class));
+        assertThat(request.includeDataStreams(), is(true));
+
+        // data streams and their backing indices should be in the authorized list
+        List<String> expectedDataStreams = List.of("logs-foo", "logs-foobar");
+        final List<String> authorizedIndices = buildAuthorizedIndices(user, SearchAction.NAME, request);
+        for (String dsName : expectedDataStreams) {
+            DataStream dataStream = metadata.dataStreams().get(dsName);
+            assertThat(authorizedIndices, hasItem(dsName));
+            for (Index i : dataStream.getIndices()) {
+                assertThat(authorizedIndices, hasItem(i.getName()));
+            }
+        }
+
+        // data streams without their backing indices will be in the resolved list since the backing indices do not match the requested
+        // pattern
+        ResolvedIndices resolvedIndices = defaultIndicesResolver.resolveIndicesAndAliases(request, metadata, authorizedIndices);
+        assertThat(resolvedIndices.getLocal(), hasItem("logs-foo"));
+        assertThat(resolvedIndices.getLocal(), hasItem("logs-foobar"));
+        assertThat(resolvedIndices.getLocal(), hasItem("logs-00001"));
+        assertThat(resolvedIndices.getLocal(), hasItem("logs-00002"));
+        assertThat(resolvedIndices.getLocal(), hasItem("logs-00003"));
+        assertThat(resolvedIndices.getLocal(), hasItem("logs-alias"));
+        for (String dsName : expectedDataStreams) {
+            DataStream dataStream = metadata.dataStreams().get(dsName);
+            assertNotNull(dataStream);
+            for (Index i : dataStream.getIndices()) {
+                assertThat(resolvedIndices.getLocal(), not(hasItem(i.getName())));
+            }
+        }
+    }
+
+    public void testDataStreamsAreVisibleWhenIncludedByRequestWithoutWildcard() {
+        final User user = new User("data-stream-tester3", "data_stream_test3");
+        String dataStreamName = "logs-foobar";
+        DataStream dataStream = metadata.dataStreams().get(dataStreamName);
+        SearchRequest request = new SearchRequest(dataStreamName);
+        assertThat(request, instanceOf(IndicesRequest.Replaceable.class));
+        assertThat(request.includeDataStreams(), is(true));
+
+        final List<String> authorizedIndices = buildAuthorizedIndices(user, SearchAction.NAME, request);
+        // data streams and their backing indices should be in the authorized list
+        assertThat(authorizedIndices, hasItem(dataStreamName));
+        for (Index i : dataStream.getIndices()) {
+            assertThat(authorizedIndices, hasItem(i.getName()));
+        }
+
+        ResolvedIndices resolvedIndices = defaultIndicesResolver.resolveIndicesAndAliases(request, metadata, authorizedIndices);
+        // data streams without their backing indices will be in the resolved list since the backing indices do not match the requested
+        // name
+        assertThat(resolvedIndices.getLocal(), hasItem(dataStreamName));
+        for (Index i : dataStream.getIndices()) {
+            assertThat(resolvedIndices.getLocal(), not(hasItem(i.getName())));
+        }
+    }
+
+    public void testBackingIndicesAreVisibleWhenIncludedByRequestWithWildcard() {
+        final User user = new User("data-stream-tester3", "data_stream_test3");
+        SearchRequest request = new SearchRequest(".ds-logs*");
+        assertThat(request, instanceOf(IndicesRequest.Replaceable.class));
+        assertThat(request.includeDataStreams(), is(true));
+
+        // data streams and their backing indices should be included in the authorized list
+        List<String> expectedDataStreams = List.of("logs-foo", "logs-foobar");
+        final List<String> authorizedIndices = buildAuthorizedIndices(user, SearchAction.NAME, request);
+        for (String dsName : expectedDataStreams) {
+            DataStream dataStream = metadata.dataStreams().get(dsName);
+            assertThat(authorizedIndices, hasItem(dsName));
+            for (Index i : dataStream.getIndices()) {
+                assertThat(authorizedIndices, hasItem(i.getName()));
+            }
+        }
+
+        // data streams should _not_ be included in the resolved list because they do not match the pattern but their backing indices
+        // should be in the resolved list because they match the pattern and are authorized via extension from their parent data stream
+        ResolvedIndices resolvedIndices = defaultIndicesResolver.resolveIndicesAndAliases(request, metadata, authorizedIndices);
+        for (String dsName : expectedDataStreams) {
+            DataStream dataStream = metadata.dataStreams().get(dsName);
+            assertThat(resolvedIndices.getLocal(), not(hasItem(dsName)));
+            for (Index i : dataStream.getIndices()) {
+                assertThat(resolvedIndices.getLocal(), hasItem(i.getName()));
+            }
+        }
+    }
+
+    public void testBackingIndicesAreNotVisibleWhenNotIncludedByRequestWithoutWildcard() {
+        final User user = new User("data-stream-tester2", "data_stream_test2");
+        String dataStreamName = "logs-foobar";
+        GetAliasesRequest request = new GetAliasesRequest(dataStreamName);
+        assertThat(request, instanceOf(IndicesRequest.Replaceable.class));
+        assertThat(request.includeDataStreams(), is(false));
+
+        // data streams and their backing indices should _not_ be in the authorized list since the backing indices
+        // did not match the requested pattern and the request does not support data streams
+        final List<String> authorizedIndices = buildAuthorizedIndices(user, GetAliasesAction.NAME, request);
+        assertThat(authorizedIndices, not(hasItem(dataStreamName)));
+        DataStream dataStream = metadata.dataStreams().get(dataStreamName);
+        assertThat(authorizedIndices, not(hasItem(dataStreamName)));
+        for (Index i : dataStream.getIndices()) {
+            assertThat(authorizedIndices, not(hasItem(i.getName())));
+        }
+
+        // neither data streams nor their backing indices will be in the resolved list since the request does not support data streams
+        // and the backing indices do not match the requested name
+        ResolvedIndices resolvedIndices = defaultIndicesResolver.resolveIndicesAndAliases(request, metadata, authorizedIndices);
+        assertThat(resolvedIndices.getLocal(), not(hasItem(dataStreamName)));
+        for (Index i : dataStream.getIndices()) {
+            assertThat(resolvedIndices.getLocal(), not(hasItem(i.getName())));
+        }
+    }
+
+    public void testDataStreamNotAuthorizedWhenBackingIndicesAreAuthorizedViaWildcardAndRequestThatIncludesDataStreams() {
+        final User user = new User("data-stream-tester2", "backing_index_test_wildcards");
+        String indexName = ".ds-logs-foobar-*";
+        SearchRequest request = new SearchRequest(indexName);
+        assertThat(request, instanceOf(IndicesRequest.Replaceable.class));
+        assertThat(request.includeDataStreams(), is(true));
+
+        // data streams should _not_ be in the authorized list but their backing indices that matched both the requested pattern
+        // and the authorized pattern should be in the list
+        final List<String> authorizedIndices = buildAuthorizedIndices(user, GetAliasesAction.NAME, request);
+        assertThat(authorizedIndices, not(hasItem("logs-foobar")));
+        DataStream dataStream = metadata.dataStreams().get("logs-foobar");
+        assertThat(authorizedIndices, not(hasItem(indexName)));
+        for (Index i : dataStream.getIndices()) {
+            assertThat(authorizedIndices, hasItem(i.getName()));
+        }
+
+        // only the backing indices will be in the resolved list since the request does not support data streams
+        // but the backing indices match the requested pattern
+        ResolvedIndices resolvedIndices = defaultIndicesResolver.resolveIndicesAndAliases(request, metadata, authorizedIndices);
+        assertThat(resolvedIndices.getLocal(), not(hasItem(dataStream.getName())));
+        for (Index i : dataStream.getIndices()) {
+            assertThat(authorizedIndices, hasItem(i.getName()));
+        }
+    }
+
+    public void testDataStreamNotAuthorizedWhenBackingIndicesAreAuthorizedViaNameAndRequestThatIncludesDataStreams() {
+        final User user = new User("data-stream-tester2", "backing_index_test_name");
+        String indexName = ".ds-logs-foobar-*";
+        SearchRequest request = new SearchRequest(indexName);
+        assertThat(request, instanceOf(IndicesRequest.Replaceable.class));
+        assertThat(request.includeDataStreams(), is(true));
+
+        // data streams should _not_ be in the authorized list but a single backing index that matched the requested pattern
+        // and the authorized name should be in the list
+        final List<String> authorizedIndices = buildAuthorizedIndices(user, GetAliasesAction.NAME, request);
+        assertThat(authorizedIndices, not(hasItem("logs-foobar")));
+        assertThat(authorizedIndices, contains(".ds-logs-foobar-000001"));
+
+        // only the single backing index will be in the resolved list since the request does not support data streams
+        // but one of the backing indices matched the requested pattern
+        ResolvedIndices resolvedIndices = defaultIndicesResolver.resolveIndicesAndAliases(request, metadata, authorizedIndices);
+        assertThat(resolvedIndices.getLocal(), not(hasItem("logs-foobar")));
+        assertThat(resolvedIndices.getLocal(), contains(".ds-logs-foobar-000001"));
+    }
+
+    public void testDataStreamNotAuthorizedWhenBackingIndicesAreAuthorizedViaWildcardAndRequestThatExcludesDataStreams() {
+        final User user = new User("data-stream-tester2", "backing_index_test_wildcards");
+        String indexName = ".ds-logs-foobar-*";
+        GetAliasesRequest request = new GetAliasesRequest(indexName);
+        assertThat(request, instanceOf(IndicesRequest.Replaceable.class));
+        assertThat(request.includeDataStreams(), is(false));
+
+        // data streams should _not_ be in the authorized list but their backing indices that matched both the requested pattern
+        // and the authorized pattern should be in the list
+        final List<String> authorizedIndices = buildAuthorizedIndices(user, GetAliasesAction.NAME, request);
+        assertThat(authorizedIndices, not(hasItem("logs-foobar")));
+        DataStream dataStream = metadata.dataStreams().get("logs-foobar");
+        assertThat(authorizedIndices, not(hasItem(indexName)));
+        for (Index i : dataStream.getIndices()) {
+            assertThat(authorizedIndices, hasItem(i.getName()));
+        }
+
+        // only the backing indices will be in the resolved list since the request does not support data streams
+        // but the backing indices match the requested pattern
+        ResolvedIndices resolvedIndices = defaultIndicesResolver.resolveIndicesAndAliases(request, metadata, authorizedIndices);
+        assertThat(resolvedIndices.getLocal(), not(hasItem(dataStream.getName())));
+        for (Index i : dataStream.getIndices()) {
+            assertThat(authorizedIndices, hasItem(i.getName()));
+        }
+    }
+
+    public void testDataStreamNotAuthorizedWhenBackingIndicesAreAuthorizedViaNameAndRequestThatExcludesDataStreams() {
+        final User user = new User("data-stream-tester2", "backing_index_test_name");
+        String indexName = ".ds-logs-foobar-*";
+        GetAliasesRequest request = new GetAliasesRequest(indexName);
+        assertThat(request, instanceOf(IndicesRequest.Replaceable.class));
+        assertThat(request.includeDataStreams(), is(false));
+
+        // data streams should _not_ be in the authorized list but a single backing index that matched the requested pattern
+        // and the authorized name should be in the list
+        final List<String> authorizedIndices = buildAuthorizedIndices(user, GetAliasesAction.NAME, request);
+        assertThat(authorizedIndices, not(hasItem("logs-foobar")));
+        assertThat(authorizedIndices, contains(".ds-logs-foobar-000001"));
+
+        // only the single backing index will be in the resolved list since the request does not support data streams
+        // but one of the backing indices matched the requested pattern
+        ResolvedIndices resolvedIndices = defaultIndicesResolver.resolveIndicesAndAliases(request, metadata, authorizedIndices);
+        assertThat(resolvedIndices.getLocal(), not(hasItem("logs-foobar")));
+        assertThat(resolvedIndices.getLocal(), contains(".ds-logs-foobar-000001"));
+    }
+
     private List<String> buildAuthorizedIndices(User user, String action) {
+        return buildAuthorizedIndices(user, action, TransportRequest.Empty.INSTANCE);
+    }
+
+    private List<String> buildAuthorizedIndices(User user, String action, TransportRequest request) {
         PlainActionFuture<Role> rolesListener = new PlainActionFuture<>();
         final Authentication authentication =
             new Authentication(user, new RealmRef("test", "indices-aliases-resolver-tests", "node"), null);
         rolesStore.getRoles(user, authentication, rolesListener);
-        return RBACEngine.resolveAuthorizedIndicesFromRole(rolesListener.actionGet(), action, metadata.getIndicesLookup());
+        return RBACEngine.resolveAuthorizedIndicesFromRole(rolesListener.actionGet(), getRequestInfo(request, action),
+            metadata.getIndicesLookup());
     }
 
     public static IndexMetadata.Builder indexBuilder(String index) {
diff --git a/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authz/RBACEngineTests.java b/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authz/RBACEngineTests.java
index dd160f2c8e6ec..8908762ca7a1d 100644
--- a/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authz/RBACEngineTests.java
+++ b/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authz/RBACEngineTests.java
@@ -11,8 +11,14 @@
 import org.elasticsearch.action.admin.cluster.stats.ClusterStatsAction;
 import org.elasticsearch.action.delete.DeleteAction;
 import org.elasticsearch.action.index.IndexAction;
+import org.elasticsearch.action.search.SearchAction;
+import org.elasticsearch.action.search.SearchRequest;
 import org.elasticsearch.action.support.PlainActionFuture;
 import org.elasticsearch.client.Client;
+import org.elasticsearch.cluster.DataStreamTestHelper;
+import org.elasticsearch.cluster.metadata.DataStream;
+import org.elasticsearch.cluster.metadata.IndexAbstraction;
+import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.collect.MapBuilder;
@@ -69,12 +75,17 @@
 import java.util.Locale;
 import java.util.Map;
 import java.util.Set;
+import java.util.TreeMap;
+import java.util.stream.Collectors;
 
 import static java.util.Collections.emptyMap;
 import static org.elasticsearch.common.util.set.Sets.newHashSet;
+import static org.elasticsearch.xpack.security.authz.AuthorizedIndicesTests.getRequestInfo;
 import static org.hamcrest.Matchers.containsInAnyOrder;
 import static org.hamcrest.Matchers.emptyIterable;
 import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.hasItem;
+import static org.hamcrest.Matchers.hasItems;
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.iterableWithSize;
@@ -1031,6 +1042,38 @@ IndexPrivilege.READ, randomBoolean(), "index-4", "index-5")
         assertThat(response.getRunAs(), containsInAnyOrder("user01", "user02"));
     }
 
+    public void testBackingIndicesAreIncludedForAuthorizedDataStreams() {
+        final String dataStreamName = "my_data_stream";
+        User user = new User(randomAlphaOfLengthBetween(4, 12));
+        Authentication authentication = mock(Authentication.class);
+        when(authentication.getUser()).thenReturn(user);
+        Role role = Role.builder("test1")
+            .cluster(Collections.singleton("all"), Collections.emptyList())
+            .add(IndexPrivilege.READ, dataStreamName)
+            .build();
+
+        TreeMap<String, IndexAbstraction> lookup = new TreeMap<>();
+        List<IndexMetadata> backingIndices = new ArrayList<>();
+        int numBackingIndices = randomIntBetween(1, 3);
+        for (int k = 0; k < numBackingIndices; k++) {
+            backingIndices.add(DataStreamTestHelper.createBackingIndex(dataStreamName, k + 1).build());
+        }
+        DataStream ds = new DataStream(dataStreamName, null,
+            backingIndices.stream().map(IndexMetadata::getIndex).collect(Collectors.toList()));
+        IndexAbstraction.DataStream iads = new IndexAbstraction.DataStream(ds, backingIndices);
+        lookup.put(ds.getName(), iads);
+        for (IndexMetadata im : backingIndices) {
+            lookup.put(im.getIndex().getName(), new IndexAbstraction.Index(im, iads));
+        }
+
+        SearchRequest request = new SearchRequest("*");
+        List<String> authorizedIndices =
+            RBACEngine.resolveAuthorizedIndicesFromRole(role, getRequestInfo(request, SearchAction.NAME), lookup);
+        assertThat(authorizedIndices, hasItem(dataStreamName));
+        assertThat(authorizedIndices, hasItems(backingIndices.stream()
+            .map(im -> im.getIndex().getName()).collect(Collectors.toList()).toArray(Strings.EMPTY_ARRAY)));
+    }
+
     private GetUserPrivilegesResponse.Indices findIndexPrivilege(Set<GetUserPrivilegesResponse.Indices> indices, String name) {
         return indices.stream().filter(i -> i.getIndices().contains(name)).findFirst().get();
     }
diff --git a/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authz/accesscontrol/IndicesPermissionTests.java b/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authz/accesscontrol/IndicesPermissionTests.java
index 9d2905e165976..e3e1967144f66 100644
--- a/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authz/accesscontrol/IndicesPermissionTests.java
+++ b/x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/authz/accesscontrol/IndicesPermissionTests.java
@@ -9,6 +9,7 @@
 import org.elasticsearch.Version;
 import org.elasticsearch.action.search.SearchAction;
 import org.elasticsearch.cluster.metadata.AliasMetadata;
+import org.elasticsearch.cluster.metadata.DataStream;
 import org.elasticsearch.cluster.metadata.IndexAbstraction;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.cluster.metadata.Metadata;
@@ -37,7 +38,9 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.SortedMap;
+import java.util.stream.Collectors;
 
+import static org.elasticsearch.cluster.DataStreamTestHelper.createTimestampField;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.is;
@@ -348,6 +351,51 @@ public void testAsyncSearchIndicesPermissions() {
         assertThat(authzMap.get(asyncSearchIndex).isGranted(), is(true));
     }
 
+    public void testAuthorizationForBackingIndices() {
+        Metadata.Builder builder = Metadata.builder();
+        String dataStreamName = randomAlphaOfLength(6);
+        int numBackingIndices = randomIntBetween(1, 3);
+        List<IndexMetadata> backingIndices = new ArrayList<>();
+        for (int backingIndexNumber = 1; backingIndexNumber <= numBackingIndices; backingIndexNumber++) {
+            backingIndices.add(createIndexMetadata(DataStream.getDefaultBackingIndexName(dataStreamName, backingIndexNumber)));
+        }
+        DataStream ds = new DataStream(dataStreamName, createTimestampField("@timestamp"),
+            backingIndices.stream().map(IndexMetadata::getIndex).collect(Collectors.toList()));
+        builder.put(ds);
+        for (IndexMetadata index : backingIndices) {
+            builder.put(index, false);
+        }
+        Metadata metadata = builder.build();
+
+        FieldPermissionsCache fieldPermissionsCache = new FieldPermissionsCache(Settings.EMPTY);
+        SortedMap<String, IndexAbstraction> lookup = metadata.getIndicesLookup();
+        IndicesPermission.Group group = new IndicesPermission.Group(IndexPrivilege.ALL, new FieldPermissions(), null, false,
+            dataStreamName);
+        Map<String, IndicesAccessControl.IndexAccessControl> authzMap = new IndicesPermission(group).authorize(
+            SearchAction.NAME,
+            Sets.newHashSet(backingIndices.stream().map(im -> im.getIndex().getName()).collect(Collectors.toList())),
+            lookup,
+            fieldPermissionsCache);
+
+        for (IndexMetadata im : backingIndices) {
+            assertThat(authzMap.get(im.getIndex().getName()).isGranted(), is(true));
+        }
+    }
+
+    private static IndexMetadata createIndexMetadata(String name) {
+        Settings.Builder settingsBuilder = Settings.builder()
+            .put(IndexMetadata.SETTING_VERSION_CREATED, Version.CURRENT)
+            .put("index.hidden", true);
+
+        IndexMetadata.Builder indexBuilder = IndexMetadata.builder(name)
+            .settings(settingsBuilder)
+            .state(IndexMetadata.State.OPEN)
+            .numberOfShards(1)
+            .numberOfReplicas(1);
+
+        return indexBuilder.build();
+    }
+
     private static FieldPermissionsDefinition fieldPermissionDef(String[] granted, String[] denied) {
         return new FieldPermissionsDefinition(granted, denied);
     }
diff --git a/x-pack/plugin/src/test/resources/rest-api-spec/test/security/authz/50_data_streams.yml b/x-pack/plugin/src/test/resources/rest-api-spec/test/security/authz/50_data_streams.yml
new file mode 100644
index 0000000000000..7c0c780dc5bfd
--- /dev/null
+++ b/x-pack/plugin/src/test/resources/rest-api-spec/test/security/authz/50_data_streams.yml
@@ -0,0 +1,149 @@
+---
+setup:
+  - skip:
+      features: ["headers", "allowed_warnings"]
+      version: " - 7.99.99"
+      reason: "change to 7.8.99 after backport"
+
+  - do:
+      cluster.health:
+        wait_for_status: yellow
+
+  - do:
+      security.put_role:
+        name: "data_stream_role"
+        body:  >
+          {
+            "indices": [
+              { "names": ["simple*"], "privileges": ["read", "write", "view_index_metadata"] }
+            ]
+          }
+
+  - do:
+      security.put_user:
+        username: "test_user"
+        body:  >
+          {
+            "password" : "x-pack-test-password",
+            "roles" : [ "data_stream_role" ],
+            "full_name" : "user with privileges on data streams but not backing indices"
+          }
+
+  - do:
+      allowed_warnings:
+        - "index template [my-template1] has index patterns [simple-data-stream1] matching patterns from existing older templates [global] with patterns (global => [*]); this template [my-template1] will take precedence during new index creation"
+      indices.put_index_template:
+        name: my-template1
+        body:
+          index_patterns: [simple-data-stream1]
+          template:
+            mappings:
+              properties:
+                '@timestamp':
+                  type: date
+          data_stream:
+            timestamp_field: '@timestamp'
+
+---
+teardown:
+  - do:
+      security.delete_user:
+        username: "test_user"
+        ignore: 404
+
+  - do:
+      security.delete_role:
+        name: "data_stream_role"
+        ignore: 404
+
+---
+"Test backing indices inherit parent data stream privileges":
+  - skip:
+      version: " - 7.99.99"
+      reason: "change to 7.8.99 after backport"
+
+  - do: # superuser
+      indices.create_data_stream:
+        name: simple-data-stream1
+  - is_true: acknowledged
+
+  - do: # superuser
+      index:
+        index:  simple-data-stream1
+        id:       1
+        op_type:  create
+        body:   { foo: bar, "@timestamp": "2020-12-12" }
+
+  - set: { _seq_no: seqno }
+  - set: { _primary_term: primary_term }
+
+  - do: # superuser
+      indices.refresh:
+        index: simple-data-stream1
+
+  # should succeed since the search request is on the data stream itself
+  - do:
+      headers: { Authorization: "Basic dGVzdF91c2VyOngtcGFjay10ZXN0LXBhc3N3b3Jk" } # test_user
+      search:
+        rest_total_hits_as_int: true
+        index: simple-data-stream1
+
+  - match: { hits.total: 1 }
+
+  # should succeed since the backing index inherits the data stream's privileges
+  - do:
+      headers: { Authorization: "Basic dGVzdF91c2VyOngtcGFjay10ZXN0LXBhc3N3b3Jk" } # test_user
+      search:
+        rest_total_hits_as_int: true
+        index: .ds-simple-data-stream1-000001
+
+  - match: { hits.total: 1 }
+
+  # should succeed since the backing index inherits the data stream's privileges
+  - do:
+      headers: { Authorization: "Basic dGVzdF91c2VyOngtcGFjay10ZXN0LXBhc3N3b3Jk" } # test_user
+      index:
+        index:  .ds-simple-data-stream1-000001
+        id: 1
+        if_seq_no: $seqno
+        if_primary_term: $primary_term
+        op_type:  index
+        body:   { foo: bar2, "@timestamp": "2020-12-12" }
+
+  - match: { _version: 2 }
+
+  - do: # superuser
+      indices.delete_data_stream:
+        name: simple-data-stream1
+  - is_true: acknowledged
+
+---
+"Test that requests not supporting data streams do not include data streams among authorized indices":
+  - skip:
+      version: " - 7.99.99"
+      reason: "change to 7.8.99 after backport"
+
+  - do: # superuser
+      indices.create_data_stream:
+        name: simple-data-stream1
+  - is_true: acknowledged
+
+  - do: # superuser
+      indices.create:
+        index: simple-index
+        body:
+          aliases:
+            simple-alias: {}
+
+  - do:
+      headers: { Authorization: "Basic dGVzdF91c2VyOngtcGFjay10ZXN0LXBhc3N3b3Jk" } # test_user
+      indices.get_alias:
+        name: simple*
+
+  - match: {simple-index.aliases.simple-alias: {}}
+  - is_false: simple-data-stream1
+
+  - do: # superuser
+      indices.delete_data_stream:
+        name: simple-data-stream1
+  - is_true: acknowledged