Respect runas realm for ApiKey security operations (elastic#52178)

When user A runs as user B and performs any API key related operations,
user B's realm should always be used to associate with the API key.
Currently user A's realm is used when getting or invalidating API keys
and owner=true. The PR is to fix this bug.

resolves: elastic#51975

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

@@ -78,22 +78,22 @@ private boolean checkIfUserIsOwnerOfApiKeys(Authentication authentication, Strin
                  * TODO bizybot we need to think on how we can propagate appropriate error message to the end user when username, realm name
                  *   is missing. This is similar to the problem of propagating right error messages in case of access denied.
                  */
-                if (authentication.getAuthenticatedBy().getType().equals(API_KEY_REALM_TYPE)) {
+                if (authentication.getSourceRealm().getType().equals(API_KEY_REALM_TYPE)) {
                     // API key cannot own any other API key so deny access
                     return false;
                 } else if (ownedByAuthenticatedUser) {
                     return true;
                 } else if (Strings.hasText(username) && Strings.hasText(realmName)) {
-                    final String authenticatedUserPrincipal = authentication.getUser().principal();
-                    final String authenticatedUserRealm = authentication.getAuthenticatedBy().getName();
-                    return username.equals(authenticatedUserPrincipal) && realmName.equals(authenticatedUserRealm);
+                    final String sourceUserPrincipal = authentication.getUser().principal();
+                    final String sourceRealmName = authentication.getSourceRealm().getName();
+                    return username.equals(sourceUserPrincipal) && realmName.equals(sourceRealmName);
                 }
             }
             return false;
         }
 
         private boolean isCurrentAuthenticationUsingSameApiKeyIdFromRequest(Authentication authentication, String apiKeyId) {
-            if (authentication.getAuthenticatedBy().getType().equals(API_KEY_REALM_TYPE)) {
+            if (authentication.getSourceRealm().getType().equals(API_KEY_REALM_TYPE)) {
                 // API key id from authentication must match the id from request
                 final String authenticatedApiKeyId = (String) authentication.getMetadata().get(API_KEY_ID_KEY);
                 if (Strings.hasText(apiKeyId)) {

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/security/authz/privilege/ManageOwnApiKeyClusterPrivilegeTests.java

@@ -97,15 +97,46 @@ public void testAuthenticationWithUserDeniesAccessToApiKeyActionsWhenItIsNotOwne
         assertFalse(clusterPermission.check("cluster:admin/xpack/security/api_key/invalidate", invalidateApiKeyRequest, authentication));
     }
 
+    public void testGetAndInvalidateApiKeyWillRespectRunAsUser() {
+        final ClusterPermission clusterPermission =
+            ManageOwnApiKeyClusterPrivilege.INSTANCE.buildPermission(ClusterPermission.builder()).build();
+
+        final Authentication authentication = createMockRunAsAuthentication(
+            "user_a", "realm_a", "realm_a_type",
+            "user_b", "realm_b", "realm_b_type");
+
+        assertTrue(clusterPermission.check("cluster:admin/xpack/security/api_key/get",
+            GetApiKeyRequest.usingRealmAndUserName("realm_b", "user_b"), authentication));
+        assertTrue(clusterPermission.check("cluster:admin/xpack/security/api_key/invalidate",
+            InvalidateApiKeyRequest.usingRealmAndUserName("realm_b", "user_b"), authentication));
+    }
+
     private Authentication createMockAuthentication(String username, String realmName, String realmType, Map<String, Object> metadata) {
         final User user = new User(username);
         final Authentication authentication = mock(Authentication.class);
         final Authentication.RealmRef authenticatedBy = mock(Authentication.RealmRef.class);
         when(authentication.getUser()).thenReturn(user);
-        when(authentication.getAuthenticatedBy()).thenReturn(authenticatedBy);
+        when(authentication.getSourceRealm()).thenReturn(authenticatedBy);
         when(authenticatedBy.getName()).thenReturn(realmName);
         when(authenticatedBy.getType()).thenReturn(realmType);
         when(authentication.getMetadata()).thenReturn(metadata);
         return authentication;
     }
+
+    private Authentication createMockRunAsAuthentication(String username, String realmName, String realmType,
+        String runAsUsername, String runAsRealmName, String runAsRealmType) {
+        final Authentication.RealmRef authenticatedBy = mock(Authentication.RealmRef.class);
+        when(authenticatedBy.getName()).thenReturn(realmName);
+        when(authenticatedBy.getType()).thenReturn(realmType);
+        final Authentication.RealmRef lookedUpBy = mock(Authentication.RealmRef.class);
+        when(lookedUpBy.getName()).thenReturn(runAsRealmName);
+        when(lookedUpBy.getType()).thenReturn(runAsRealmType);
+        final User user = new User(runAsUsername, new String[0], new User(username));
+        final Authentication authentication = mock(Authentication.class);
+        when(authentication.getUser()).thenReturn(user);
+        when(authentication.getAuthenticatedBy()).thenReturn(authenticatedBy);
+        when(authentication.getSourceRealm()).thenReturn(lookedUpBy);
+        when(authentication.getMetadata()).thenReturn(Collections.emptyMap());
+        return authentication;
+    }
 }

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

@@ -892,7 +892,22 @@ public static String getCreatorRealmName(final Authentication authentication) {
         if (authentication.getAuthenticatedBy().getType().equals(API_KEY_REALM_TYPE)) {
             return (String) authentication.getMetadata().get(API_KEY_CREATOR_REALM_NAME);
         } else {
-            return authentication.getAuthenticatedBy().getName();
+            return authentication.getSourceRealm().getName();
+        }
+    }
+
+    /**
+     * Returns realm type for the authenticated user.
+     * If the user is authenticated by realm type {@value API_KEY_REALM_TYPE}
+     * then it will return the realm name of user who created this API key.
+     * @param authentication {@link Authentication}
+     * @return realm type
+     */
+    public static String getCreatorRealmType(final Authentication authentication) {
+        if (authentication.getAuthenticatedBy().getType().equals(API_KEY_REALM_TYPE)) {
+            return (String) authentication.getMetadata().get(API_KEY_CREATOR_REALM_TYPE);
+        } else {
+            return authentication.getSourceRealm().getType();
         }
     }
 

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/ingest/SetSecurityUserProcessor.java

@@ -114,17 +114,11 @@ public IngestDocument execute(IngestDocument ingestDocument) throws Exception {
                     final Map<String, Object> realmField =
                         existingRealmField instanceof Map ? (Map<String, Object>) existingRealmField : new HashMap<>();
 
-                    final Object realmName, realmType;
-                    if (Authentication.AuthenticationType.API_KEY == authentication.getAuthenticationType()) {
-                        realmName = authentication.getMetadata().get(ApiKeyService.API_KEY_CREATOR_REALM_NAME);
-                        realmType = authentication.getMetadata().get(ApiKeyService.API_KEY_CREATOR_REALM_TYPE);
-                    } else {
-                        realmName = authentication.getSourceRealm().getName();
-                        realmType = authentication.getSourceRealm().getType();
-                    }
+                    final Object realmName = ApiKeyService.getCreatorRealmName(authentication);
                     if (realmName != null) {
                         realmField.put("name", realmName);
                     }
+                    final Object realmType = ApiKeyService.getCreatorRealmType(authentication);
                     if (realmType != null) {
                         realmField.put("type", realmType);
                     }