Ensure getUser() is the logical user, not API key creator for RCS 2.0 (…

…#107023)

This commit changes SecurityContext#getUser() to provide the original user that 
initiated the call when run across clusters for RCS 2.0. Before this change the getUser() 
would provide the RCS 2.0 API key creator as the current user.

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

@@ -35,6 +35,7 @@
 import java.util.function.Consumer;
 import java.util.function.Function;
 
+import static org.elasticsearch.xpack.core.security.authc.Authentication.getAuthenticationFromCrossClusterAccessMetadata;
 import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.AUTHENTICATION_KEY;
 import static org.elasticsearch.xpack.core.security.authz.AuthorizationServiceField.AUTHORIZATION_INFO_KEY;
 
@@ -71,6 +72,11 @@ public User requireUser() {
     @Nullable
     public User getUser() {
         Authentication authentication = getAuthentication();
+        if (authentication != null) {
+            if (authentication.isCrossClusterAccess()) {
+                authentication = getAuthenticationFromCrossClusterAccessMetadata(authentication);
+            }
+        }
         return authentication == null ? null : authentication.getEffectiveSubject().getUser();
     }
 

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

@@ -24,6 +24,7 @@
 import org.elasticsearch.xpack.core.security.authc.Authentication.RealmRef;
 import org.elasticsearch.xpack.core.security.authc.AuthenticationField;
 import org.elasticsearch.xpack.core.security.authc.AuthenticationTestHelper;
+import org.elasticsearch.xpack.core.security.authc.CrossClusterAccessSubjectInfo;
 import org.elasticsearch.xpack.core.security.authz.AuthorizationEngine.AuthorizationInfo;
 import org.elasticsearch.xpack.core.security.authz.AuthorizationEngine.ParentActionAuthorization;
 import org.elasticsearch.xpack.core.security.authz.AuthorizationServiceField;
@@ -78,6 +79,20 @@ public void testGetAuthenticationAndUser() throws IOException {
         assertEquals(user, securityContext.getUser());
     }
 
+    public void testGetUserForAPIKeyBasedCrossCluster() throws IOException {
+        final User user = new User("test");
+        final CrossClusterAccessSubjectInfo crossClusterAccessSubjectInfo = AuthenticationTestHelper.randomCrossClusterAccessSubjectInfo(
+            AuthenticationTestHelper.builder().user(user).realmRef(new RealmRef("ldap", "foo", "node1")).build(false)
+        );
+        final Authentication authentication = AuthenticationTestHelper.builder()
+            .crossClusterAccess(randomAlphaOfLengthBetween(10, 20), crossClusterAccessSubjectInfo)
+            .build(false);
+        User apiKeyUser = authentication.getEffectiveSubject().getUser();
+        authentication.writeToContext(threadContext);
+        assertEquals(user, securityContext.getUser());
+        assertNotEquals(apiKeyUser, securityContext.getUser());
+    }
+
     public void testGetAuthenticationDoesNotSwallowIOException() {
         threadContext.putHeader(AuthenticationField.AUTHENTICATION_KEY, ""); // an intentionally corrupt header
         final SecurityContext securityContext = new SecurityContext(Settings.EMPTY, threadContext);