Respect runas realm for ApiKey security operations

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

@@ -96,15 +96,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(Map.of());
+        return authentication;
+    }
 }

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

@@ -884,7 +884,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);
                     }