Improve user check for resource sharing and its documentation (#69844)

The user check when accessing shared resources is improved to be more accurate. 
Also add documentation about how it works and its limitations.

docs/reference/search/async-search.asciidoc

@@ -139,8 +139,8 @@ set to `false`.
 
 The get async search API retrieves the results of a previously submitted
 async search request given its id. If the {es} {security-features} are enabled,
-the access to the results of a specific async search is restricted to the user
-that submitted it in the first place.
+the access to the results of a specific async search is restricted to
+<<can-access-resources-check,the user or API key that submitted it>>.
 
 [source,console,id=get-async-search-date-histogram-example]
 --------------------------------------------------

docs/reference/search/scroll-api.asciidoc

@@ -69,7 +69,9 @@ parameter indicates how long {es} should retain the
 
 The search response returns a scroll ID in the `_scroll_id` response body
 parameter. You can then use the scroll ID with the scroll API to retrieve the
-next batch of results for the request.
+next batch of results for the request. If the {es} {security-features} are enabled,
+the access to the results of a specific scroll ID is restricted to
+<<can-access-resources-check,the user or API key that submitted the search>>.
 
 You can also use the scroll API to specify a new `scroll` parameter that extends
 or shortens the retention period for the search context.

x-pack/docs/en/security/limitations.asciidoc

@@ -90,3 +90,27 @@ LDAP Groups.  For example, if a user is a member of `group_1` and `group_1` is a
 member of `group_2`, only `group_1` will be discovered. However, the
 <<active-directory-realm, Active Directory Realm>> *does* support transitive
 group membership.
+
+
+[discrete]
+[[can-access-resources-check]]
+=== Resource sharing check for users and API keys
+
+The result of <<async-search,async search>> and <<scroll-api,scroll>> requests can be retrieved later
+by the same user or API key that submitted the initial request. The verification process involves comparing
+the username, authentication realm type, and (for realms other than file or native) realm name.
+If you used an API key to submit the request, only that key can retrieve the results.
+This logic also has a few limitations:
+
+* Two different realms can have the same name on different nodes. This is not a
+recommended way to configure realms, therefore the resource sharing check
+does not attempt to detect this inconsistency.
+* Realms can be renamed. This can cause inconsistency for the resource sharing check
+when you submit an async search or scroll then rename the realm and try to retrieve the results.
+Hence, changing realm names should be handled with care since it can cause complications for more than
+just the resource sharing check.
+* The username is dynamically computed for realms backed by certain external authentication
+providers. For example, the username can be derived from part of the DN in an LDAP realm.
+It is in theory possible that two distinct users from the external system get
+mapped to the same username. Our recommendation is to avoid this situation in the first place.
+Hence, the resource sharing check does not account for this potential discrepancy.

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/async/AsyncTaskIndexService.java

@@ -459,28 +459,7 @@ public final class AsyncTaskIndexService<R extends AsyncResponse<R>> {
             return false;
         }
         Authentication origin = AuthenticationContextSerializer.decode(originHeaders.get(AUTHENTICATION_KEY));
-        return ensureAuthenticatedUserIsSame(origin, current);
-    }
-
-    /**
-     * Compares the {@link Authentication} that was used to create the {@link AsyncExecutionId} with the
-     * current authentication.
-     */
-    boolean ensureAuthenticatedUserIsSame(Authentication original, Authentication current) {
-        final boolean samePrincipal = original.getUser().principal().equals(current.getUser().principal());
-        final boolean sameRealmType;
-        if (original.getUser().isRunAs()) {
-            if (current.getUser().isRunAs()) {
-                sameRealmType = original.getLookedUpBy().getType().equals(current.getLookedUpBy().getType());
-            }  else {
-                sameRealmType = original.getLookedUpBy().getType().equals(current.getAuthenticatedBy().getType());
-            }
-        } else if (current.getUser().isRunAs()) {
-            sameRealmType = original.getAuthenticatedBy().getType().equals(current.getLookedUpBy().getType());
-        } else {
-            sameRealmType = original.getAuthenticatedBy().getType().equals(current.getAuthenticatedBy().getType());
-        }
-        return samePrincipal && sameRealmType;
+        return origin.canAccessResourcesOf(current);
     }
 
     /**

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authc/Authentication.java

@@ -14,6 +14,8 @@ import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.common.xcontent.ToXContentObject;
 import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.xpack.core.security.authc.esnative.NativeRealmSettings;
+import org.elasticsearch.xpack.core.security.authc.file.FileRealmSettings;
 import org.elasticsearch.xpack.core.security.authc.support.AuthenticationContextSerializer;
 import org.elasticsearch.xpack.core.security.user.InternalUserSerializationHelper;
 import org.elasticsearch.xpack.core.security.user.User;
@@ -21,10 +23,13 @@ import org.elasticsearch.xpack.core.security.user.User;
 import java.io.IOException;
 import java.util.Base64;
 import java.util.Collections;
+import java.util.EnumSet;
 import java.util.Locale;
 import java.util.Map;
 import java.util.Objects;
 
+import static org.elasticsearch.xpack.core.security.authz.privilege.ManageOwnApiKeyClusterPrivilege.API_KEY_ID_KEY;
+
 // TODO(hub-cap) Clean this up after moving User over - This class can re-inherit its field AUTHENTICATION_KEY in AuthenticationField.
 // That interface can be removed
 public class Authentication implements ToXContentObject {
@@ -130,6 +135,57 @@ public class Authentication implements ToXContentObject {
         out.writeMap(metadata);
     }
 
+    /**
+     * Checks whether the user or API key of the passed in authentication can access the resources owned by the user
+     * or API key of this authentication. The rules are as follows:
+     *   * True if the authentications are for the same API key (same API key ID)
+     *   * True if they are the same username from the same realm
+     *      - For file and native realm, same realm means the same realm type
+     *      - For all other realms, same realm means same realm type plus same realm name
+     *   * An user and its API key cannot access each other's resources
+     *   * An user and its token can access each other's resources
+     *   * Two API keys are never able to access each other's resources regardless of their ownership.
+     *
+     *  This check is a best effort and it does not account for certain static and external changes.
+     *  See also <a href="https://www.elastic.co/guide/en/elasticsearch/reference/master/security-limitations.html">
+     *      security limitations</a>
+     */
+    public boolean canAccessResourcesOf(Authentication other) {
+        if (AuthenticationType.API_KEY == getAuthenticationType() && AuthenticationType.API_KEY == other.getAuthenticationType()) {
+            final boolean sameKeyId = getMetadata().get(API_KEY_ID_KEY).equals(other.getMetadata().get(API_KEY_ID_KEY));
+            if (sameKeyId) {
+                assert getUser().principal().equals(other.getUser().principal()) :
+                    "The same API key ID cannot be attributed to two different usernames";
+            }
+            return sameKeyId;
+        }
+
+        if (getAuthenticationType().equals(other.getAuthenticationType())
+            || (AuthenticationType.REALM == getAuthenticationType() && AuthenticationType.TOKEN == other.getAuthenticationType())
+            || (AuthenticationType.TOKEN == getAuthenticationType() && AuthenticationType.REALM == other.getAuthenticationType())) {
+            if (false == getUser().principal().equals(other.getUser().principal())) {
+                return false;
+            }
+            final RealmRef thisRealm = getSourceRealm();
+            final RealmRef otherRealm = other.getSourceRealm();
+            if (FileRealmSettings.TYPE.equals(thisRealm.getType()) || NativeRealmSettings.TYPE.equals(thisRealm.getType())) {
+                return thisRealm.getType().equals(otherRealm.getType());
+            }
+            return thisRealm.getName().equals(otherRealm.getName()) && thisRealm.getType().equals(otherRealm.getType());
+        } else {
+            assert EnumSet.of(
+                AuthenticationType.REALM,
+                AuthenticationType.API_KEY,
+                AuthenticationType.TOKEN,
+                AuthenticationType.ANONYMOUS,
+                AuthenticationType.INTERNAL
+            ).containsAll(EnumSet.of(getAuthenticationType(), other.getAuthenticationType()))
+                : "cross AuthenticationType comparison for canAccessResourcesOf is not applicable for: "
+                + EnumSet.of(getAuthenticationType(), other.getAuthenticationType());
+            return false;
+        }
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) return true;

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authz/privilege/ManageOwnApiKeyClusterPrivilege.java

@@ -25,7 +25,7 @@ import java.util.Arrays;
 public class ManageOwnApiKeyClusterPrivilege implements NamedClusterPrivilege {
     public static final ManageOwnApiKeyClusterPrivilege INSTANCE = new ManageOwnApiKeyClusterPrivilege();
     private static final String PRIVILEGE_NAME = "manage_own_api_key";
-    private static final String API_KEY_ID_KEY = "_security_api_key_id";
+    public static final String API_KEY_ID_KEY = "_security_api_key_id";
 
     private final ClusterPermission permission;
 

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/async/AsyncTaskServiceTests.java

@@ -80,7 +80,7 @@ public class AsyncTaskServiceTests extends ESSingleNodeTestCase {
             new Authentication(new User("test", "role"), new Authentication.RealmRef("realm", "file", "node"), null);
         Authentication current = randomBoolean() ? original :
             new Authentication(new User("test", "role"), new Authentication.RealmRef("realm", "file", "node"), null);
-        assertTrue(indexService.ensureAuthenticatedUserIsSame(original, current));
+        assertTrue(original.canAccessResourcesOf(current));
         ThreadContext threadContext = new ThreadContext(Settings.EMPTY);
         original.writeToContext(threadContext);
         assertTrue(indexService.ensureAuthenticatedUserIsSame(threadContext.getHeaders(), current));
@@ -94,14 +94,14 @@ public class AsyncTaskServiceTests extends ESSingleNodeTestCase {
         User user = new User(new User("test", "role"), new User("authenticated", "runas"));
         current = new Authentication(user, new Authentication.RealmRef("realm", "file", "node"),
             new Authentication.RealmRef(randomAlphaOfLengthBetween(1, 16), "file", "node"));
-        assertTrue(indexService.ensureAuthenticatedUserIsSame(original, current));
+        assertTrue(original.canAccessResourcesOf(current));
         assertTrue(indexService.ensureAuthenticatedUserIsSame(threadContext.getHeaders(), current));
 
         // both user are run as
         current = new Authentication(user, new Authentication.RealmRef("realm", "file", "node"),
             new Authentication.RealmRef(randomAlphaOfLengthBetween(1, 16), "file", "node"));
         Authentication runAs = current;
-        assertTrue(indexService.ensureAuthenticatedUserIsSame(runAs, current));
+        assertTrue(runAs.canAccessResourcesOf(current));
         threadContext = new ThreadContext(Settings.EMPTY);
         original.writeToContext(threadContext);
         assertTrue(indexService.ensureAuthenticatedUserIsSame(threadContext.getHeaders(), current));
@@ -111,24 +111,24 @@ public class AsyncTaskServiceTests extends ESSingleNodeTestCase {
             new Authentication(new User("test", "role"), new Authentication.RealmRef("realm", randomAlphaOfLength(5), "node"), null);
         threadContext = new ThreadContext(Settings.EMPTY);
         original.writeToContext(threadContext);
-        assertFalse(indexService.ensureAuthenticatedUserIsSame(original, differentRealmType));
+        assertFalse(original.canAccessResourcesOf(differentRealmType));
         assertFalse(indexService.ensureAuthenticatedUserIsSame(threadContext.getHeaders(), differentRealmType));
 
         // wrong user
         Authentication differentUser =
             new Authentication(new User("test2", "role"), new Authentication.RealmRef("realm", "realm", "node"), null);
-        assertFalse(indexService.ensureAuthenticatedUserIsSame(original, differentUser));
+        assertFalse(original.canAccessResourcesOf(differentUser));
 
         // run as different user
         Authentication diffRunAs = new Authentication(new User(new User("test2", "role"), new User("authenticated", "runas")),
             new Authentication.RealmRef("realm", "file", "node1"), new Authentication.RealmRef("realm", "file", "node1"));
-        assertFalse(indexService.ensureAuthenticatedUserIsSame(original, diffRunAs));
+        assertFalse(original.canAccessResourcesOf(diffRunAs));
         assertFalse(indexService.ensureAuthenticatedUserIsSame(threadContext.getHeaders(), diffRunAs));
 
         // run as different looked up by type
         Authentication runAsDiffType = new Authentication(user, new Authentication.RealmRef("realm", "file", "node"),
             new Authentication.RealmRef(randomAlphaOfLengthBetween(1, 16), randomAlphaOfLengthBetween(5, 12), "node"));
-        assertFalse(indexService.ensureAuthenticatedUserIsSame(original, runAsDiffType));
+        assertFalse(original.canAccessResourcesOf(runAsDiffType));
         assertFalse(indexService.ensureAuthenticatedUserIsSame(threadContext.getHeaders(), runAsDiffType));
     }
 

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/security/authc/AuthenticationTests.java

@@ -7,14 +7,28 @@
 
 package org.elasticsearch.xpack.core.security.authc;
 
+import org.elasticsearch.Version;
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.test.VersionUtils;
+import org.elasticsearch.xpack.core.security.authc.Authentication.AuthenticationType;
+import org.elasticsearch.xpack.core.security.authc.Authentication.RealmRef;
+import org.elasticsearch.xpack.core.security.authc.esnative.NativeRealmSettings;
+import org.elasticsearch.xpack.core.security.authc.file.FileRealmSettings;
 import org.elasticsearch.xpack.core.security.user.User;
 
+import java.util.Arrays;
+import java.util.EnumSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import static org.elasticsearch.xpack.core.security.authz.privilege.ManageOwnApiKeyClusterPrivilege.API_KEY_ID_KEY;
+
 public class AuthenticationTests extends ESTestCase {
 
     public void testWillGetLookedUpByWhenItExists() {
-        final Authentication.RealmRef authenticatedBy = new Authentication.RealmRef("auth_by", "auth_by_type", "node");
-        final Authentication.RealmRef lookedUpBy = new Authentication.RealmRef("lookup_by", "lookup_by_type", "node");
+        final RealmRef authenticatedBy = new RealmRef("auth_by", "auth_by_type", "node");
+        final RealmRef lookedUpBy = new RealmRef("lookup_by", "lookup_by_type", "node");
         final Authentication authentication = new Authentication(
             new User("user"), authenticatedBy, lookedUpBy);
 
@@ -22,11 +36,134 @@ public class AuthenticationTests extends ESTestCase {
     }
 
     public void testWillGetAuthenticateByWhenLookupIsNull() {
-        final Authentication.RealmRef authenticatedBy = new Authentication.RealmRef("auth_by", "auth_by_type", "node");
+        final RealmRef authenticatedBy = new RealmRef("auth_by", "auth_by_type", "node");
         final Authentication authentication = new Authentication(
             new User("user"), authenticatedBy, null);
 
         assertEquals(authenticatedBy, authentication.getSourceRealm());
     }
 
+    public void testCanAccessResourcesOf() {
+        // Same user is the same
+        final User user1 = randomUser();
+        final RealmRef realm1 = randomRealm();
+        checkCanAccessResources(randomAuthentication(user1, realm1), randomAuthentication(user1, realm1));
+
+        // Different username is different no matter which realm it is from
+        final User user2 = randomValueOtherThanMany(u -> u.principal().equals(user1.principal()), this::randomUser);
+        // user 2 can be from either the same realm or a different realm
+        final RealmRef realm2 = randomFrom(realm1, randomRealm());
+        assertCannotAccessResources(randomAuthentication(user1, realm2), randomAuthentication(user2, realm2));
+
+        // Same username but different realm is different
+        final RealmRef realm3;
+        switch (randomIntBetween(0, 2)) {
+            case 0: // change name
+                realm3 = mutateRealm(realm1, randomAlphaOfLengthBetween(3, 8), null);
+                if (realmIsSingleton(realm1)) {
+                    checkCanAccessResources(randomAuthentication(user1, realm1), randomAuthentication(user1, realm3));
+                } else {
+                    assertCannotAccessResources(randomAuthentication(user1, realm1), randomAuthentication(user1, realm3));
+                }
+                break;
+            case 1: // change type
+                realm3 = mutateRealm(realm1, null, randomAlphaOfLengthBetween(3, 8));
+                assertCannotAccessResources(randomAuthentication(user1, realm1), randomAuthentication(user1, realm3));
+                break;
+            case 2: // both
+                realm3 = mutateRealm(realm1, randomAlphaOfLengthBetween(3, 8), randomAlphaOfLengthBetween(3, 8));
+                assertCannotAccessResources(randomAuthentication(user1, realm1), randomAuthentication(user1, realm3));
+                break;
+            default:
+                assert false : "Case number out of range";
+        }
+
+        // User and its API key are not the same owner
+        assertCannotAccessResources(randomAuthentication(user1, realm1),
+            randomApiKeyAuthentication(user1, randomAlphaOfLengthBetween(10, 20)));
+
+        // Same API key ID are the same owner
+        final String apiKeyId1 = randomAlphaOfLengthBetween(10, 20);
+        checkCanAccessResources(randomApiKeyAuthentication(user1, apiKeyId1), randomApiKeyAuthentication(user1, apiKeyId1));
+
+        // Two API keys (2 API key IDs) are not the same owner
+        final String apiKeyId2 = randomValueOtherThan(apiKeyId1, () -> randomAlphaOfLengthBetween(10, 20));
+        assertCannotAccessResources(randomApiKeyAuthentication(randomFrom(user1, user2), apiKeyId1),
+            randomApiKeyAuthentication(randomFrom(user1, user2), apiKeyId2));
+    }
+
+    private void checkCanAccessResources(Authentication authentication0, Authentication authentication1) {
+        if (authentication0.getAuthenticationType() == authentication1.getAuthenticationType()
+            || EnumSet.of(AuthenticationType.REALM, AuthenticationType.TOKEN).equals(
+                EnumSet.of(authentication0.getAuthenticationType(), authentication1.getAuthenticationType()))) {
+            assertTrue(authentication0.canAccessResourcesOf(authentication1));
+            assertTrue(authentication1.canAccessResourcesOf(authentication0));
+        } else {
+            assertCannotAccessResources(authentication0, authentication1);
+        }
+    }
+
+    private void assertCannotAccessResources(Authentication authentication0, Authentication authentication1) {
+        assertFalse(authentication0.canAccessResourcesOf(authentication1));
+        assertFalse(authentication1.canAccessResourcesOf(authentication0));
+    }
+
+    private User randomUser() {
+        return new User(randomAlphaOfLengthBetween(3, 8),
+            randomArray(1, 3, String[]::new, () -> randomAlphaOfLengthBetween(3, 8)));
+    }
+
+    private RealmRef randomRealm() {
+        return new RealmRef(
+            randomAlphaOfLengthBetween(3, 8),
+            randomFrom(FileRealmSettings.TYPE, NativeRealmSettings.TYPE, randomAlphaOfLengthBetween(3, 8)),
+            randomAlphaOfLengthBetween(3, 8));
+    }
+
+    private RealmRef mutateRealm(RealmRef original, String name, String type) {
+        return new RealmRef(
+            name == null ? original.getName() : name,
+            type == null ? original.getType() : type,
+            randomBoolean() ? original.getNodeName() : randomAlphaOfLengthBetween(3, 8));
+    }
+
+    private Authentication randomAuthentication(User user, RealmRef realmRef) {
+        if (user == null) {
+            user = new User(randomAlphaOfLengthBetween(3, 8),
+                randomArray(1, 3, String[]::new, () -> randomAlphaOfLengthBetween(3, 8)));
+        }
+        if (realmRef == null) {
+            realmRef = randomRealm();
+        }
+        final Version version = VersionUtils.randomVersionBetween(random(), Version.V_7_0_0, Version.CURRENT);
+        final AuthenticationType authenticationType =
+            randomValueOtherThan(AuthenticationType.API_KEY, () -> randomFrom(AuthenticationType.values()));
+        final Map<String, Object> metadata;
+        if (randomBoolean()) {
+            metadata = Map.of(randomAlphaOfLengthBetween(3, 8), randomAlphaOfLengthBetween(3, 8));
+        } else {
+            metadata = Arrays.stream(randomArray(1, 5, String[]::new, () -> randomAlphaOfLengthBetween(3, 8)))
+                .collect(Collectors.toMap(s -> s, s -> randomAlphaOfLengthBetween(3, 8)));
+        }
+        if (randomBoolean()) { // run-as
+            return new Authentication(new User(user.principal(), user.roles(), randomUser()),
+                randomRealm(), realmRef, version, authenticationType, metadata);
+        } else {
+            return new Authentication(user, realmRef, null, version, authenticationType, metadata);
+        }
+    }
+
+    private Authentication randomApiKeyAuthentication(User user, String apiKeyId) {
+        final RealmRef apiKeyRealm = new RealmRef("_es_api_key", "_es_api_key", randomAlphaOfLengthBetween(3, 8));
+        return new Authentication(user,
+            apiKeyRealm,
+            null,
+            VersionUtils.randomVersionBetween(random(), Version.V_7_0_0, Version.CURRENT),
+            AuthenticationType.API_KEY,
+            Map.of(API_KEY_ID_KEY, apiKeyId));
+    }
+
+    private boolean realmIsSingleton(RealmRef realmRef) {
+        return Set.of(FileRealmSettings.TYPE, NativeRealmSettings.TYPE).contains(realmRef.getType());
+    }
 }

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

@@ -123,23 +123,7 @@ public final class SecuritySearchOperationListener implements SearchOperationLis
     static void ensureAuthenticatedUserIsSame(Authentication original, Authentication current, AuditTrailService auditTrailService,
                                               ShardSearchContextId id, String action, TransportRequest request, String requestId,
                                               AuthorizationInfo authorizationInfo) {
-        // this is really a best effort attempt since we cannot guarantee principal uniqueness
-        // and realm names can change between nodes.
-        final boolean samePrincipal = original.getUser().principal().equals(current.getUser().principal());
-        final boolean sameRealmType;
-        if (original.getUser().isRunAs()) {
-            if (current.getUser().isRunAs()) {
-                sameRealmType = original.getLookedUpBy().getType().equals(current.getLookedUpBy().getType());
-            } else {
-                sameRealmType = original.getLookedUpBy().getType().equals(current.getAuthenticatedBy().getType());
-            }
-        } else if (current.getUser().isRunAs()) {
-            sameRealmType = original.getAuthenticatedBy().getType().equals(current.getLookedUpBy().getType());
-        } else {
-            sameRealmType = original.getAuthenticatedBy().getType().equals(current.getAuthenticatedBy().getType());
-        }
-
-        final boolean sameUser = samePrincipal && sameRealmType;
+        final boolean sameUser = original.canAccessResourcesOf(current);
         if (sameUser == false) {
             auditTrailService.get().accessDenied(requestId, current, action, request, authorizationInfo);
             throw new SearchContextMissingException(id);