Check whether index is same and only one write index exists for alias

When same alias points to multiple indices we can write to only one index
with `is_write_index` value `true`. The special handling for
PutMappingRequest filtered out such aliases making the request unauthorized.
The check has been modified to consider write index flag and only when the
requested index matches with the one with write index alias.

Closes elastic#40831

Author: Yogesh Gaikwad

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

@@ -248,8 +248,17 @@ static String getPutMappingIndexOrAlias(PutMappingRequest request, List<String>
             if (aliasMetaData != null) {
                 Optional<String> foundAlias = aliasMetaData.stream()
                     .map(AliasMetaData::alias)
+                    .filter(aliasName -> {
+                        List<IndexMetaData> indexMetadata = metaData.getAliasAndIndexLookup().get(aliasName).getIndices();
+                        if (indexMetadata.size() == 1) {
+                            return true;
+                        } else {
+                            IndexMetaData idxMeta = ((AliasOrIndex.Alias) metaData.getAliasAndIndexLookup().get(aliasName))
+                                    .getWriteIndex();
+                            return idxMeta != null && idxMeta.getIndex().getName().equals(concreteIndexName);
+                        }
+                    })
                     .filter(authorizedIndicesList::contains)
-                    .filter(aliasName -> metaData.getAliasAndIndexLookup().get(aliasName).getIndices().size() == 1)
                     .findFirst();
                 resolvedAliasOrIndex = foundAlias.orElse(concreteIndexName);
             } else {

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

@@ -72,6 +72,7 @@
 import org.junit.Before;
 
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -138,6 +139,9 @@ public void setup() {
                 .put(indexBuilder("-index11").settings(settings))
                 .put(indexBuilder("-index20").settings(settings))
                 .put(indexBuilder("-index21").settings(settings))
+                .put(indexBuilder("logs-00001").putAlias(AliasMetaData.builder("logs-alias").writeIndex(false)).settings(settings))
+                .put(indexBuilder("logs-00002").putAlias(AliasMetaData.builder("logs-alias").writeIndex(false)).settings(settings))
+                .put(indexBuilder("logs-00003").putAlias(AliasMetaData.builder("logs-alias").writeIndex(true)).settings(settings))
                 .put(indexBuilder(securityIndexName).settings(settings)).build();
 
         if (withAlias) {
@@ -1355,6 +1359,23 @@ public void testDynamicPutMappingRequestFromAlias() {
         request = new PutMappingRequest(Strings.EMPTY_ARRAY).setConcreteIndex(new Index(index, UUIDs.base64UUID()));
         putMappingIndexOrAlias = IndicesAndAliasesResolver.getPutMappingIndexOrAlias(request, authorizedIndices, metaData);
         assertEquals(index, putMappingIndexOrAlias);
+
+    }
+
+    public void testDynamicPutMappingRequestForOneAliasForMultipleIndices() {
+        String index = "logs-00003";
+        PutMappingRequest request = new PutMappingRequest(Strings.EMPTY_ARRAY).setConcreteIndex(new Index(index, UUIDs.base64UUID()));
+        List<String> authorizedIndices = Collections.singletonList("logs-alias");
+        assert metaData.getAliasAndIndexLookup().get("logs-alias").getIndices().size() == 3;
+        String putMappingIndexOrAlias = IndicesAndAliasesResolver.getPutMappingIndexOrAlias(request, authorizedIndices, metaData);
+        assertEquals("logs-alias", putMappingIndexOrAlias);
+
+        index = "logs-00002";
+        request = new PutMappingRequest(Strings.EMPTY_ARRAY).setConcreteIndex(new Index(index, UUIDs.base64UUID()));
+        authorizedIndices = Collections.singletonList("logs-alias");
+        assert metaData.getAliasAndIndexLookup().get("logs-alias").getIndices().size() == 3;
+        putMappingIndexOrAlias = IndicesAndAliasesResolver.getPutMappingIndexOrAlias(request, authorizedIndices, metaData);
+        assertEquals(index, putMappingIndexOrAlias);
     }
 
     // TODO with the removal of DeleteByQuery is there another way to test resolving a write action?

x-pack/plugin/src/test/resources/rest-api-spec/test/security/authz/32_ilm_using_alias.yml

@@ -0,0 +1,179 @@
+---
+
+setup:
+  - skip:
+      features: headers
+
+  - do:
+      cluster.health:
+          wait_for_status: yellow
+
+  - do:
+      security.put_role:
+        name: "foo_alias_role"
+        body:  >
+            {
+              "indices": [
+                { "names": ["foo_alias"], "privileges": ["write", "manage"] }
+              ]
+            }
+
+  - do:
+      security.put_user:
+        username: "test_user"
+        body:  >
+          {
+            "password" : "x-pack-test-password",
+            "roles" : [ "foo_alias_role" ],
+            "full_name" : "user with privileges to write, manage via alias"
+          }
+
+  - do:
+      cluster.put_settings:
+        body:  >
+          {
+            "transient" : { "indices.lifecycle.poll_interval" : "1s" }
+          }
+
+  - do:
+      ilm.put_lifecycle:
+        policy: "foo-policy"
+        body:  >
+          {
+            "policy" : {
+               "phases" : {
+                  "hot" : {
+                    "min_age": "0ms",
+                    "actions" : {
+                       "rollover" : {
+                          "max_docs" : 2
+                       }
+                    }
+                  }
+               }
+             }
+          }
+
+  - do:
+      indices.put_template:
+        name: "foo-template"
+        body:  >
+          {
+            "index_patterns": ["foo-logs-*"],
+            "settings": {
+              "number_of_shards": 1,
+              "number_of_replicas": 0,
+              "index.lifecycle.name": "foo-policy",
+              "index.lifecycle.rollover_alias": "foo_alias"
+            }
+          }
+
+  - do:
+      indices.create:
+        index: foo-logs-000001
+        body:  >
+          {
+            "aliases": {
+              "foo_alias" : {  }
+            }
+          }
+
+---
+teardown:
+  - do:
+      security.delete_user:
+        username: "test_user"
+        ignore: 404
+
+  - do:
+      security.delete_role:
+        name: "foo_alias_role"
+        ignore: 404
+
+  - do:
+      indices.delete_alias:
+        index: "foo-logs-*"
+        name: [ "foo_alias" ]
+        ignore: 404
+
+  - do:
+      indices.delete:
+        index: [ "foo-logs-*" ]
+        ignore: 404
+
+---
+"Test ilm, index via write alias of index":
+
+  # index using alias
+  - do:
+      headers: { Authorization: "Basic dGVzdF91c2VyOngtcGFjay10ZXN0LXBhc3N3b3Jk" } # test_user
+      create:
+        id: 1
+        index: foo_alias
+        body: >
+          {
+            "name" : "doc1"
+          }
+
+  - do:
+      headers: { Authorization: "Basic dGVzdF91c2VyOngtcGFjay10ZXN0LXBhc3N3b3Jk" } # test_user
+      create:
+        id: 2
+        index: foo_alias
+        body: >
+          {
+            "name" : "doc2"
+          }
+
+  - do:
+      indices.refresh: {}
+
+  # sleep so ilm task gets triggered
+  # this is a hacky way to sleep for 1s, since we will never have 10 nodes
+  - do:
+      catch: request_timeout
+      cluster.health:
+        wait_for_nodes: 10
+        timeout: "2s"
+  - match:
+     timed_out: true
+
+  # ILM should have triggered rollover by now
+
+  # ensure new index is created
+  - do:
+      indices.exists:
+        index: foo-logs-000002
+
+  - is_true: ''
+
+  # index using alias
+  - do:
+      headers: { Authorization: "Basic dGVzdF91c2VyOngtcGFjay10ZXN0LXBhc3N3b3Jk" } # test_user
+      create:
+        id: 3
+        index: foo_alias
+        body: >
+          {
+            "name" : "doc3"
+          }
+
+  - do:
+      indices.refresh: {}
+
+  # verify the indexes
+  - do:
+      search:
+        rest_total_hits_as_int: true
+        index: foo-logs-000001
+
+  - match: { hits.total: 2 }
+  - match: { hits.hits.0._index: "foo-logs-000001"}
+
+  - do:
+      search:
+        rest_total_hits_as_int: true
+        index: foo-logs-000002
+
+  - match: { hits.total: 1 }
+  - match: { hits.hits.0._index: "foo-logs-000002"}