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

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: #51975

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

@@ -78,22 +78,22 @@ public class ManageOwnApiKeyClusterPrivilege implements NamedClusterPrivilege {
                  * 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 sourceUserPrincipal = authentication.getUser().principal();
-                    final String authenticatedUserRealm = authentication.getAuthenticatedBy().getName();
+                    final String sourceRealmName = authentication.getSourceRealm().getName();
-                    return username.equals(authenticatedUserPrincipal) && realmName.equals(authenticatedUserRealm);
+                    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 class ManageOwnApiKeyClusterPrivilegeTests extends ESTestCase {
         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 class ApiKeyService {
         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 final class SetSecurityUserProcessor extends AbstractProcessor {
                     final Map<String, Object> realmField =
                         existingRealmField instanceof Map ? (Map<String, Object>) existingRealmField : new HashMap<>();
 
-                    final Object realmName, realmType;
+                    final Object realmName = ApiKeyService.getCreatorRealmName(authentication);
-                    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();
-                    }
                     if (realmName != null) {
                         realmField.put("name", realmName);
                     }
+                    final Object realmType = ApiKeyService.getCreatorRealmType(authentication);
                     if (realmType != null) {
                         realmField.put("type", realmType);
                     }

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

@@ -16,6 +16,8 @@ import org.elasticsearch.action.support.WriteRequest;
 import org.elasticsearch.action.update.UpdateResponse;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.collect.MapBuilder;
+import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.rest.RestStatus;
@@ -30,6 +32,8 @@ import org.elasticsearch.xpack.core.security.action.GetApiKeyRequest;
 import org.elasticsearch.xpack.core.security.action.GetApiKeyResponse;
 import org.elasticsearch.xpack.core.security.action.InvalidateApiKeyRequest;
 import org.elasticsearch.xpack.core.security.action.InvalidateApiKeyResponse;
+import org.elasticsearch.xpack.core.security.action.user.PutUserRequest;
+import org.elasticsearch.xpack.core.security.action.user.PutUserResponse;
 import org.elasticsearch.xpack.core.security.authc.support.UsernamePasswordToken;
 import org.elasticsearch.xpack.core.security.authz.RoleDescriptor;
 import org.elasticsearch.xpack.core.security.client.SecurityClient;
@@ -45,6 +49,7 @@ import java.util.Arrays;
 import java.util.Base64;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
@@ -95,7 +100,9 @@ public class ApiKeyIntegTests extends SecurityIntegTestCase {
             "manage_api_key_role:\n" +
             "  cluster: [\"manage_api_key\"]\n" +
             "manage_own_api_key_role:\n" +
-            "  cluster: [\"manage_own_api_key\"]\n";
+            "  cluster: [\"manage_own_api_key\"]\n" +
+            "run_as_role:\n" +
+            "  run_as: [\"user_with_manage_own_api_key_role\"]\n";
     }
 
     @Override
@@ -121,7 +128,7 @@ public class ApiKeyIntegTests extends SecurityIntegTestCase {
             assertBusy(() -> assertFalse(apiKeyService.isExpirationInProgress()));
         }
     }
-    
+
     public void testCreateApiKey() {
         // Get an instant without nanoseconds as the expiration has millisecond precision
         final Instant start = Instant.ofEpochMilli(Instant.now().toEpochMilli());
@@ -546,6 +553,54 @@ public class ApiKeyIntegTests extends SecurityIntegTestCase {
             response, userWithManageApiKeyRoleApiKeys.stream().map(o -> o.getId()).collect(Collectors.toSet()), null);
     }
 
+    public void testGetApiKeysOwnedByRunAsUserWhenOwnerIsTrue() throws ExecutionException, InterruptedException {
+        createUserWithRunAsRole();
+        int noOfSuperuserApiKeys = randomIntBetween(3, 5);
+        int noOfApiKeysForUserWithManageApiKeyRole = randomIntBetween(3, 5);
+        createApiKeys(noOfSuperuserApiKeys, null);
+        List<CreateApiKeyResponse> userWithManageOwnApiKeyRoleApiKeys = createApiKeys("user_with_manage_own_api_key_role",
+            "user_with_run_as_role", noOfApiKeysForUserWithManageApiKeyRole, null, "monitor");
+        PlainActionFuture<GetApiKeyResponse> listener = new PlainActionFuture<>();
+        new SecurityClient(getClientForRunAsUser()).getApiKey(GetApiKeyRequest.forOwnedApiKeys(), listener);
+        GetApiKeyResponse response = listener.get();
+        verifyGetResponse("user_with_manage_own_api_key_role", noOfApiKeysForUserWithManageApiKeyRole, userWithManageOwnApiKeyRoleApiKeys,
+            response, userWithManageOwnApiKeyRoleApiKeys.stream().map(o -> o.getId()).collect(Collectors.toSet()), null);
+    }
+
+    public void testGetApiKeysOwnedByRunAsUserWhenRunAsUserInfoIsGiven() throws ExecutionException, InterruptedException {
+        createUserWithRunAsRole();
+        int noOfSuperuserApiKeys = randomIntBetween(3, 5);
+        int noOfApiKeysForUserWithManageApiKeyRole = randomIntBetween(3, 5);
+        createApiKeys(noOfSuperuserApiKeys, null);
+        List<CreateApiKeyResponse> userWithManageOwnApiKeyRoleApiKeys = createApiKeys("user_with_manage_own_api_key_role",
+            "user_with_run_as_role", noOfApiKeysForUserWithManageApiKeyRole, null, "monitor");
+        PlainActionFuture<GetApiKeyResponse> listener = new PlainActionFuture<>();
+        new SecurityClient(getClientForRunAsUser()).getApiKey(
+            GetApiKeyRequest.usingRealmAndUserName("file", "user_with_manage_own_api_key_role"), listener);
+        GetApiKeyResponse response = listener.get();
+        verifyGetResponse("user_with_manage_own_api_key_role", noOfApiKeysForUserWithManageApiKeyRole, userWithManageOwnApiKeyRoleApiKeys,
+            response, userWithManageOwnApiKeyRoleApiKeys.stream().map(o -> o.getId()).collect(Collectors.toSet()), null);
+    }
+
+    public void testGetApiKeysOwnedByRunAsUserWillNotWorkWhenAuthUserInfoIsGiven() throws ExecutionException, InterruptedException {
+        createUserWithRunAsRole();
+        int noOfSuperuserApiKeys = randomIntBetween(3, 5);
+        int noOfApiKeysForUserWithManageApiKeyRole = randomIntBetween(3, 5);
+        createApiKeys(noOfSuperuserApiKeys, null);
+        final List<CreateApiKeyResponse> userWithManageOwnApiKeyRoleApiKeys = createApiKeys("user_with_manage_own_api_key_role",
+            "user_with_run_as_role", noOfApiKeysForUserWithManageApiKeyRole, null, "monitor");
+        PlainActionFuture<GetApiKeyResponse> listener = new PlainActionFuture<>();
+        final Tuple<String,String> invalidRealmAndUserPair = randomFrom(
+            new Tuple<>("file", "user_with_run_as_role"),
+            new Tuple<>("index", "user_with_manage_own_api_key_role"),
+            new Tuple<>("index", "user_with_run_as_role"));
+        new SecurityClient(getClientForRunAsUser()).getApiKey(
+            GetApiKeyRequest.usingRealmAndUserName(invalidRealmAndUserPair.v1(), invalidRealmAndUserPair.v2()), listener);
+        final ElasticsearchSecurityException e = expectThrows(ElasticsearchSecurityException.class, listener::actionGet);
+        assertThat(e.getMessage(), containsString(
+            "unauthorized for user [user_with_run_as_role] run as [user_with_manage_own_api_key_role]"));
+    }
+
     public void testGetAllApiKeys() throws InterruptedException, ExecutionException {
         int noOfSuperuserApiKeys = randomIntBetween(3, 5);
         int noOfApiKeysForUserWithManageApiKeyRole = randomIntBetween(3, 5);
@@ -609,6 +664,53 @@ public class ApiKeyIntegTests extends SecurityIntegTestCase {
         verifyInvalidateResponse(noOfApiKeysForUserWithManageApiKeyRole, userWithManageApiKeyRoleApiKeys, invalidateResponse);
     }
 
+    public void testInvalidateApiKeysOwnedByRunAsUserWhenOwnerIsTrue() throws InterruptedException, ExecutionException {
+        createUserWithRunAsRole();
+        int noOfSuperuserApiKeys = randomIntBetween(3, 5);
+        int noOfApiKeysForUserWithManageApiKeyRole = randomIntBetween(3, 5);
+        createApiKeys(noOfSuperuserApiKeys, null);
+        List<CreateApiKeyResponse> userWithManageApiKeyRoleApiKeys = createApiKeys("user_with_manage_own_api_key_role",
+            "user_with_run_as_role", noOfApiKeysForUserWithManageApiKeyRole, null, "monitor");
+        PlainActionFuture<InvalidateApiKeyResponse> listener = new PlainActionFuture<>();
+        new SecurityClient(getClientForRunAsUser()).invalidateApiKey(
+            InvalidateApiKeyRequest.forOwnedApiKeys(), listener);
+        InvalidateApiKeyResponse invalidateResponse = listener.get();
+        verifyInvalidateResponse(noOfApiKeysForUserWithManageApiKeyRole, userWithManageApiKeyRoleApiKeys, invalidateResponse);
+    }
+
+    public void testInvalidateApiKeysOwnedByRunAsUserWhenRunAsUserInfoIsGiven() throws InterruptedException, ExecutionException {
+        createUserWithRunAsRole();
+        int noOfSuperuserApiKeys = randomIntBetween(3, 5);
+        int noOfApiKeysForUserWithManageApiKeyRole = randomIntBetween(3, 5);
+        createApiKeys(noOfSuperuserApiKeys, null);
+        List<CreateApiKeyResponse> userWithManageApiKeyRoleApiKeys = createApiKeys("user_with_manage_own_api_key_role",
+            "user_with_run_as_role", noOfApiKeysForUserWithManageApiKeyRole, null, "monitor");
+        PlainActionFuture<InvalidateApiKeyResponse> listener = new PlainActionFuture<>();
+        new SecurityClient(getClientForRunAsUser()).invalidateApiKey(
+            InvalidateApiKeyRequest.usingRealmAndUserName("file", "user_with_manage_own_api_key_role"), listener);
+        InvalidateApiKeyResponse invalidateResponse = listener.get();
+        verifyInvalidateResponse(noOfApiKeysForUserWithManageApiKeyRole, userWithManageApiKeyRoleApiKeys, invalidateResponse);
+    }
+
+    public void testInvalidateApiKeysOwnedByRunAsUserWillNotWorkWhenAuthUserInfoIsGiven() throws InterruptedException, ExecutionException {
+        createUserWithRunAsRole();
+        int noOfSuperuserApiKeys = randomIntBetween(3, 5);
+        int noOfApiKeysForUserWithManageApiKeyRole = randomIntBetween(3, 5);
+        createApiKeys(noOfSuperuserApiKeys, null);
+        List<CreateApiKeyResponse> userWithManageApiKeyRoleApiKeys = createApiKeys("user_with_manage_own_api_key_role",
+            "user_with_run_as_role", noOfApiKeysForUserWithManageApiKeyRole, null, "monitor");
+        PlainActionFuture<InvalidateApiKeyResponse> listener = new PlainActionFuture<>();
+        final Tuple<String,String> invalidRealmAndUserPair = randomFrom(
+            new Tuple<>("file", "user_with_run_as_role"),
+            new Tuple<>("index", "user_with_manage_own_api_key_role"),
+            new Tuple<>("index", "user_with_run_as_role"));
+        new SecurityClient(getClientForRunAsUser()).invalidateApiKey(
+            InvalidateApiKeyRequest.usingRealmAndUserName(invalidRealmAndUserPair.v1(), invalidRealmAndUserPair.v2()), listener);
+        final ElasticsearchSecurityException e = expectThrows(ElasticsearchSecurityException.class, listener::actionGet);
+        assertThat(e.getMessage(), containsString(
+            "unauthorized for user [user_with_run_as_role] run as [user_with_manage_own_api_key_role]"));
+    }
+
     public void testApiKeyAuthorizationApiKeyMustBeAbleToRetrieveItsOwnInformationButNotAnyOtherKeysCreatedBySameOwner()
         throws InterruptedException, ExecutionException {
         List<CreateApiKeyResponse> responses = createApiKeys(SecuritySettingsSource.TEST_SUPERUSER,2, null, (String[]) null);
@@ -710,15 +812,32 @@ public class ApiKeyIntegTests extends SecurityIntegTestCase {
     }
 
     private List<CreateApiKeyResponse> createApiKeys(String user, int noOfApiKeys, TimeValue expiration, String... clusterPrivileges) {
+        final Map<String, String> headers = Collections.singletonMap(
+                "Authorization", UsernamePasswordToken.basicAuthHeaderValue(user, SecuritySettingsSourceField.TEST_PASSWORD_SECURE_STRING));
+        return createApiKeys(headers, noOfApiKeys, expiration, clusterPrivileges);
+    }
+
+    private List<CreateApiKeyResponse> createApiKeys(String owningUser, String authenticatingUser,
+        int noOfApiKeys, TimeValue expiration, String... clusterPrivileges) {
+        final Map<String, String> headers = new MapBuilder<String, String>()
+            .put("Authorization",
+                UsernamePasswordToken.basicAuthHeaderValue(
+                    authenticatingUser, SecuritySettingsSourceField.TEST_PASSWORD_SECURE_STRING))
+            .put("es-security-runas-user", owningUser)
+            .immutableMap();
+        return createApiKeys(headers, noOfApiKeys, expiration, clusterPrivileges);
+    }
+
+    private List<CreateApiKeyResponse> createApiKeys(Map<String, String> headers,
+        int noOfApiKeys, TimeValue expiration, String... clusterPrivileges) {
         List<CreateApiKeyResponse> responses = new ArrayList<>();
         for (int i = 0; i < noOfApiKeys; i++) {
             final RoleDescriptor descriptor = new RoleDescriptor("role", clusterPrivileges, null, null);
-            Client client = client().filterWithHeader(Collections.singletonMap("Authorization", UsernamePasswordToken
+            Client client = client().filterWithHeader(headers);
-                    .basicAuthHeaderValue(user, SecuritySettingsSourceField.TEST_PASSWORD_SECURE_STRING)));
             SecurityClient securityClient = new SecurityClient(client);
             final CreateApiKeyResponse response = securityClient.prepareCreateApiKey()
-                    .setName("test-key-" + randomAlphaOfLengthBetween(5, 9) + i).setExpiration(expiration)
+                .setName("test-key-" + randomAlphaOfLengthBetween(5, 9) + i).setExpiration(expiration)
-                    .setRoleDescriptors(Collections.singletonList(descriptor)).get();
+                .setRoleDescriptors(Collections.singletonList(descriptor)).get();
             assertNotNull(response.getId());
             assertNotNull(response.getKey());
             responses.add(response);
@@ -727,6 +846,38 @@ public class ApiKeyIntegTests extends SecurityIntegTestCase {
         return responses;
     }
 
+    /**
+     * In order to have negative tests for realm name mismatch, user_with_run_as_role
+     * needs to be created in a different realm other than file (which is handled by configureUsers()).
+     * This new helper method creates the user in the native realm.
+     */
+    private void createUserWithRunAsRole() throws ExecutionException, InterruptedException {
+        final PutUserRequest putUserRequest = new PutUserRequest();
+        putUserRequest.username("user_with_run_as_role");
+        putUserRequest.roles("run_as_role");
+        putUserRequest.passwordHash(SecuritySettingsSource.TEST_PASSWORD_HASHED.toCharArray());
+        PlainActionFuture<PutUserResponse> listener = new PlainActionFuture<>();
+        final Map<String, String> headers = new MapBuilder<String, String>()
+            .put("Authorization",
+                UsernamePasswordToken.basicAuthHeaderValue(
+                    SecuritySettingsSource.TEST_SUPERUSER, SecuritySettingsSourceField.TEST_PASSWORD_SECURE_STRING))
+            .immutableMap();
+        final Client client = client().filterWithHeader(headers);
+        new SecurityClient(client).putUser(putUserRequest, listener);
+        final PutUserResponse putUserResponse = listener.get();
+        assertTrue(putUserResponse.created());
+    }
+
+    private Client getClientForRunAsUser() {
+        final Map<String, String> headers = new MapBuilder<String, String>()
+            .put("Authorization",
+                UsernamePasswordToken.basicAuthHeaderValue(
+                    "user_with_run_as_role", SecuritySettingsSourceField.TEST_PASSWORD_SECURE_STRING))
+            .put("es-security-runas-user", "user_with_manage_own_api_key_role")
+            .immutableMap();
+        return client().filterWithHeader(headers);
+    }
+
     private void assertErrorMessage(final ElasticsearchSecurityException ese, String action, String userName, String apiKeyId) {
         assertThat(ese.getMessage(),
             is("action [" + action + "] is unauthorized for API key id [" + apiKeyId + "] of user [" + userName + "]"));

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

@@ -566,12 +566,20 @@ public class ApiKeyServiceTests extends ESTestCase {
         assertNull(cachedApiKeyHashResult);
     }
 
-    public void testWillAlwaysGetAuthenticationRealmName() {
+    public void testWillGetLookedUpByRealmNameIfExists() {
         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 Authentication.RealmRef lookedUpBy = new Authentication.RealmRef("looked_up_by", "looked_up_by_type", "node");
         final Authentication authentication = new Authentication(
             new User("user"), authenticatedBy, lookedUpBy);
-        assertEquals("auth_by", ApiKeyService.getCreatorRealmName(authentication));
+        assertEquals("looked_up_by", ApiKeyService.getCreatorRealmName(authentication));
+    }
+
+    public void testWillGetLookedUpByRealmTypeIfExists() {
+        final Authentication.RealmRef authenticatedBy = new Authentication.RealmRef("auth_by", "auth_by_type", "node");
+        final Authentication.RealmRef lookedUpBy = new Authentication.RealmRef("looked_up_by", "looked_up_by_type", "node");
+        final Authentication authentication = new Authentication(
+            new User("user"), authenticatedBy, lookedUpBy);
+        assertEquals("looked_up_by_type", ApiKeyService.getCreatorRealmType(authentication));
     }
 
     private ApiKeyService createApiKeyService(Settings baseSettings) {

x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/ingest/SetSecurityUserProcessorTests.java

@@ -68,9 +68,9 @@ public class SetSecurityUserProcessorTests extends ESTestCase {
         User user = Mockito.mock(User.class);
         Authentication authentication = Mockito.mock(Authentication.class);
         Mockito.when(authentication.getUser()).thenReturn(user);
-        final Authentication.RealmRef authRealm = new Authentication.RealmRef("_name", "_type", "_node_name");
+        final Authentication.RealmRef authByRealm = new Authentication.RealmRef("_name", "_type", "_node_name");
-        Mockito.when(authentication.getAuthenticatedBy()).thenReturn(authRealm);
+        Mockito.when(authentication.getSourceRealm()).thenReturn(authByRealm);
-        Mockito.when(authentication.getSourceRealm()).thenReturn(authRealm);
+        Mockito.when(authentication.getAuthenticatedBy()).thenReturn(authByRealm);
         Mockito.when(authentication.getAuthenticationType()).thenReturn(AuthenticationType.REALM);
         Mockito.when(authentication.encode()).thenReturn(randomAlphaOfLength(24)); // don't care as long as it's not null
         new AuthenticationContextSerializer().writeToContext(authentication, threadContext);