Internalise run-as user and simplify the regular User class (#86246)

Today the run-as information is spread between User (authenticatedUser) and
Authentication (lookup realm). They have to be configured consistently to work
correctly. Previously there was also no inherent logic to ensure the
consistency. Recent refactoring of Authentication class has made the situation
better by favour Authentication creation with dedicated convenient methods over
free-for-all constructors. #86206 is the ongoing PR that will finally remove
public access of Authentication constructors. Now that the Authentication class
is being tightly controlled, it makes possible to clean-up the User class.
Specifically, the run-as information is already provided by the Authentication
class, there is no need for the User class to also keep track of it. In fact,
the way User class tracking the authenticating user with an inner user is less
straightforward and not friendly to serialisation. Also, conceptually run-as is
an information at Authentication level instead of User level.

This PR refactors the User class so that it no longer keeps track of run-as
information so that there is a single consistent way to check whether an
Authentication object is run-as. The essential changes are:

* Removes User.isRunAs() method 
* Removes User.authenticatedUser() method and its backing instance variable 
* Removes all User constructors that take authenticatedUser as an argument 
* Adds a new private RunAsUser class inside Authentication 

Note that this RunAsUser class is not really necessary in long term, the plan
is to remove it in the later refactoring of Authentication class (where Subject
variable and methods will become primary over the current User).  It is added
mostly to make the refactoring easier and reduce the change to how things work
conceptually. This class is not exposed and there should be no need to use this
class outside of Authentication itself. This tight scope should make it
relatively easy to remove it later.

Relates: #86206 
Relates: #80117

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/action/user/GetUsersResponse.java

@@ -9,6 +9,7 @@
 import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.xpack.core.security.authc.Authentication;
 import org.elasticsearch.xpack.core.security.user.User;
 
 import java.io.IOException;
@@ -19,7 +20,7 @@
  */
 public class GetUsersResponse extends ActionResponse {
 
-    private User[] users;
+    private final User[] users;
 
     public GetUsersResponse(StreamInput in) throws IOException {
         super(in);
@@ -29,7 +30,9 @@ public GetUsersResponse(StreamInput in) throws IOException {
         } else {
             users = new User[size];
             for (int i = 0; i < size; i++) {
-                users[i] = User.readFrom(in);
+                final User user = Authentication.AuthenticationSerializationHelper.readUserFrom(in);
+                assert false == User.isInternal(user) : "should not get internal users";
+                users[i] = user;
             }
         }
     }
@@ -55,7 +58,7 @@ public void writeTo(StreamOutput out) throws IOException {
         out.writeVInt(users == null ? -1 : users.length);
         if (users != null) {
             for (User user : users) {
-                User.writeTo(user, out);
+                Authentication.AuthenticationSerializationHelper.writeUserTo(user, out);
             }
         }
     }

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

@@ -10,6 +10,7 @@
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.Assertions;
 import org.elasticsearch.Version;
+import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.io.stream.BytesStreamOutput;
 import org.elasticsearch.common.io.stream.StreamInput;
@@ -29,15 +30,20 @@
 import org.elasticsearch.xpack.core.security.authc.service.ServiceAccountSettings;
 import org.elasticsearch.xpack.core.security.authc.support.AuthenticationContextSerializer;
 import org.elasticsearch.xpack.core.security.user.AnonymousUser;
-import org.elasticsearch.xpack.core.security.user.InternalUserSerializationHelper;
+import org.elasticsearch.xpack.core.security.user.AsyncSearchUser;
+import org.elasticsearch.xpack.core.security.user.SecurityProfileUser;
+import org.elasticsearch.xpack.core.security.user.SystemUser;
 import org.elasticsearch.xpack.core.security.user.User;
+import org.elasticsearch.xpack.core.security.user.XPackSecurityUser;
+import org.elasticsearch.xpack.core.security.user.XPackUser;
 
 import java.io.IOException;
 import java.io.UncheckedIOException;
 import java.util.Base64;
 import java.util.Collections;
 import java.util.EnumSet;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 import java.util.Objects;
@@ -51,6 +57,8 @@
 import static org.elasticsearch.xpack.core.security.authc.Authentication.RealmRef.newServiceAccountRealmRef;
 import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.ANONYMOUS_REALM_NAME;
 import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.ANONYMOUS_REALM_TYPE;
+import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.API_KEY_REALM_NAME;
+import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.API_KEY_REALM_TYPE;
 import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.ATTACH_REALM_NAME;
 import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.ATTACH_REALM_TYPE;
 import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.FALLBACK_REALM_NAME;
@@ -90,8 +98,8 @@ private Authentication(
         this.version = version;
         this.type = type;
         this.metadata = metadata;
-        if (user.isRunAs()) {
-            authenticatingSubject = new Subject(user.authenticatedUser(), authenticatedBy, version, metadata);
+        if (user instanceof RunAsUser runAsUser) {
+            authenticatingSubject = new Subject(runAsUser.authenticatingUser, authenticatedBy, version, metadata);
             // The lookup user for run-as currently doesn't have authentication metadata associated with them because
             // lookupUser only returns the User object. The lookup user for authorization delegation does have
             // authentication metadata, but the realm does not expose this difference between authenticatingUser and
@@ -105,7 +113,7 @@ private Authentication(
     }
 
     public Authentication(StreamInput in) throws IOException {
-        this.user = InternalUserSerializationHelper.readFrom(in);
+        this.user = AuthenticationSerializationHelper.readUserFrom(in);
         this.authenticatedBy = new RealmRef(in);
         if (in.readBoolean()) {
             this.lookedUpBy = new RealmRef(in);
@@ -115,8 +123,8 @@ public Authentication(StreamInput in) throws IOException {
         this.version = in.getVersion();
         type = AuthenticationType.values()[in.readVInt()];
         metadata = in.readMap();
-        if (user.isRunAs()) {
-            authenticatingSubject = new Subject(user.authenticatedUser(), authenticatedBy, version, metadata);
+        if (user instanceof RunAsUser runAsUser) {
+            authenticatingSubject = new Subject(runAsUser.authenticatingUser, authenticatedBy, version, metadata);
             // The lookup user for run-as currently doesn't have authentication metadata associated with them because
             // lookupUser only returns the User object. The lookup user for authorization delegation does have
             // authentication metadata, but the realm does not expose this difference between authenticatingUser and
@@ -235,11 +243,13 @@ public Authentication maybeRewriteForOlderVersion(Version olderVersion) {
      */
     public Authentication runAs(User runAs, @Nullable RealmRef lookupRealmRef) {
         Objects.requireNonNull(runAs);
-        assert false == runAs.isRunAs();
-        assert false == getUser().isRunAs();
+        assert false == runAs instanceof RunAsUser;
+        assert false == runAs instanceof AnonymousUser;
+        assert false == getUser() instanceof RunAsUser;
+        assert false == hasSyntheticRealmNameOrType(lookupRealmRef) : "should not use synthetic realm name/type for lookup realms";
         assert AuthenticationType.REALM == getAuthenticationType() || AuthenticationType.API_KEY == getAuthenticationType();
         return new Authentication(
-            new User(runAs, getUser()),
+            new RunAsUser(runAs, getUser()),
             getAuthenticatedBy(),
             lookupRealmRef,
             getVersion(),
@@ -297,24 +307,31 @@ public Authentication maybeAddAnonymousRoles(@Nullable AnonymousUser anonymousUs
         }
         final String[] allRoleNames = ArrayUtils.concat(getUser().roles(), anonymousUser.roles());
 
-        return new Authentication(
-            new User(
+        final User user;
+        if (getUser()instanceof RunAsUser runAsUser) {
+            user = new RunAsUser(
                 new User(
-                    getUser().principal(),
+                    runAsUser.principal(),
                     allRoleNames,
-                    getUser().fullName(),
-                    getUser().email(),
-                    getUser().metadata(),
-                    getUser().enabled()
+                    runAsUser.fullName(),
+                    runAsUser.email(),
+                    runAsUser.metadata(),
+                    runAsUser.enabled()
                 ),
-                getUser().authenticatedUser()
-            ),
-            getAuthenticatedBy(),
-            getLookedUpBy(),
-            getVersion(),
-            getAuthenticationType(),
-            getMetadata()
-        );
+                runAsUser.authenticatingUser
+            );
+        } else {
+            user = new User(
+                getUser().principal(),
+                allRoleNames,
+                getUser().fullName(),
+                getUser().email(),
+                getUser().metadata(),
+                getUser().enabled()
+            );
+        }
+
+        return new Authentication(user, getAuthenticatedBy(), getLookedUpBy(), getVersion(), getAuthenticationType(), getMetadata());
     }
 
     /**
@@ -388,7 +405,7 @@ public String encode() throws IOException {
     }
 
     public void writeTo(StreamOutput out) throws IOException {
-        InternalUserSerializationHelper.writeTo(user, out);
+        AuthenticationSerializationHelper.writeUserTo(user, out);
         authenticatedBy.writeTo(out);
         if (lookedUpBy != null) {
             out.writeBoolean(true);
@@ -528,6 +545,21 @@ private void assertDomainAssignment() {
         }
     }
 
+    private boolean hasSyntheticRealmNameOrType(@Nullable RealmRef realmRef) {
+        if (realmRef == null) {
+            return false;
+        }
+        if (List.of(API_KEY_REALM_NAME, ServiceAccountSettings.REALM_NAME, ANONYMOUS_REALM_NAME, FALLBACK_REALM_NAME, ATTACH_REALM_NAME)
+            .contains(realmRef.getName())) {
+            return true;
+        }
+        if (List.of(API_KEY_REALM_TYPE, ServiceAccountSettings.REALM_TYPE, ANONYMOUS_REALM_TYPE, FALLBACK_REALM_TYPE, ATTACH_REALM_TYPE)
+            .contains(realmRef.getType())) {
+            return true;
+        }
+        return false;
+    }
+
     @Override
     public String toString() {
         StringBuilder builder = new StringBuilder("Authentication[").append(user)
@@ -671,7 +703,7 @@ static RealmRef newServiceAccountRealmRef(String nodeName) {
 
         static RealmRef newApiKeyRealmRef(String nodeName) {
             // no domain for API Key tokens
-            return new RealmRef(AuthenticationField.API_KEY_REALM_NAME, AuthenticationField.API_KEY_REALM_TYPE, nodeName, null);
+            return new RealmRef(API_KEY_REALM_NAME, API_KEY_REALM_TYPE, nodeName, null);
         }
     }
 
@@ -740,7 +772,7 @@ public static Authentication newAnonymousAuthentication(AnonymousUser anonymousU
 
     public static Authentication newServiceAccountAuthentication(User serviceAccountUser, String nodeName, Map<String, Object> metadata) {
         // TODO make the service account user a separate class/interface
-        assert false == serviceAccountUser.isRunAs();
+        assert false == serviceAccountUser instanceof RunAsUser;
         final Authentication.RealmRef authenticatedBy = newServiceAccountRealmRef(nodeName);
         Authentication authentication = new Authentication(
             serviceAccountUser,
@@ -756,7 +788,7 @@ public static Authentication newServiceAccountAuthentication(User serviceAccount
 
     public static Authentication newRealmAuthentication(User user, RealmRef realmRef) {
         // TODO make the type system ensure that this is not a run-as user
-        assert false == user.isRunAs();
+        assert false == user instanceof RunAsUser;
         Authentication authentication = new Authentication(user, realmRef, null, Version.CURRENT, AuthenticationType.REALM, Map.of());
         assert false == authentication.isServiceAccount();
         assert false == authentication.isApiKey();
@@ -768,7 +800,7 @@ public static Authentication newRealmAuthentication(User user, RealmRef realmRef
     public static Authentication newApiKeyAuthentication(AuthenticationResult<User> authResult, String nodeName) {
         assert authResult.isAuthenticated() : "API Key authn result must be successful";
         final User apiKeyUser = authResult.getValue();
-        assert false == apiKeyUser.isRunAs();
+        assert false == apiKeyUser instanceof RunAsUser;
         assert apiKeyUser.roles().length == 0 : "The user associated to an API key authentication must have no role";
         final Authentication.RealmRef authenticatedBy = newApiKeyRealmRef(nodeName);
         Authentication authentication = new Authentication(
@@ -868,4 +900,127 @@ public enum AuthenticationType {
         ANONYMOUS,
         INTERNAL
     }
+
+    // Package private for testing
+    static class RunAsUser extends User {
+        final User authenticatingUser;
+
+        RunAsUser(User effectiveUser, User authenticatingUser) {
+            super(
+                effectiveUser.principal(),
+                effectiveUser.roles(),
+                effectiveUser.fullName(),
+                effectiveUser.email(),
+                effectiveUser.metadata(),
+                effectiveUser.enabled()
+            );
+            this.authenticatingUser = Objects.requireNonNull(authenticatingUser);
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) return true;
+            if (o == null || getClass() != o.getClass()) return false;
+            if (false == super.equals(o)) return false;
+            RunAsUser runAsUser = (RunAsUser) o;
+            return authenticatingUser.equals(runAsUser.authenticatingUser);
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(super.hashCode(), authenticatingUser);
+        }
+
+        @Override
+        public String toString() {
+            StringBuilder sb = new StringBuilder();
+            sb.append("RunAsUser[username=").append(principal());
+            sb.append(",roles=[").append(Strings.arrayToCommaDelimitedString(roles())).append("]");
+            sb.append(",fullName=").append(fullName());
+            sb.append(",email=").append(email());
+            sb.append(",metadata=");
+            sb.append(metadata());
+            if (enabled() == false) {
+                sb.append(",(disabled)");
+            }
+            sb.append(",authenticatingUser=[").append(authenticatingUser.toString()).append("]");
+            sb.append("]");
+            return sb.toString();
+        }
+    }
+
+    public static class AuthenticationSerializationHelper {
+
+        private AuthenticationSerializationHelper() {}
+
+        public static User readUserFrom(StreamInput input) throws IOException {
+            final boolean isInternalUser = input.readBoolean();
+            final String username = input.readString();
+            if (isInternalUser) {
+                if (SystemUser.is(username)) {
+                    return SystemUser.INSTANCE;
+                } else if (XPackUser.is(username)) {
+                    return XPackUser.INSTANCE;
+                } else if (XPackSecurityUser.is(username)) {
+                    return XPackSecurityUser.INSTANCE;
+                } else if (SecurityProfileUser.is(username)) {
+                    return SecurityProfileUser.INSTANCE;
+                } else if (AsyncSearchUser.is(username)) {
+                    return AsyncSearchUser.INSTANCE;
+                }
+                throw new IllegalStateException("user [" + username + "] is not an internal user");
+            }
+            return partialReadUserFrom(username, input);
+        }
+
+        public static void writeUserTo(User user, StreamOutput output) throws IOException {
+            if (SystemUser.is(user)) {
+                output.writeBoolean(true);
+                output.writeString(SystemUser.NAME);
+            } else if (XPackUser.is(user)) {
+                output.writeBoolean(true);
+                output.writeString(XPackUser.NAME);
+            } else if (XPackSecurityUser.is(user)) {
+                output.writeBoolean(true);
+                output.writeString(XPackSecurityUser.NAME);
+            } else if (SecurityProfileUser.is(user)) {
+                output.writeBoolean(true);
+                output.writeString(SecurityProfileUser.NAME);
+            } else if (AsyncSearchUser.is(user)) {
+                output.writeBoolean(true);
+                output.writeString(AsyncSearchUser.NAME);
+            } else {
+                doWriteUserTo(user, output);
+            }
+        }
+
+        private static User partialReadUserFrom(String username, StreamInput input) throws IOException {
+            String[] roles = input.readStringArray();
+            Map<String, Object> metadata = input.readMap();
+            String fullName = input.readOptionalString();
+            String email = input.readOptionalString();
+            boolean enabled = input.readBoolean();
+            User outerUser = new User(username, roles, fullName, email, metadata, enabled);
+            boolean hasInnerUser = input.readBoolean();
+            if (hasInnerUser) {
+                User innerUser = readUserFrom(input);
+                assert false == User.isInternal(innerUser) : "authenticating user cannot be internal";
+                return new RunAsUser(outerUser, innerUser);
+            } else {
+                return outerUser;
+            }
+        }
+
+        private static void doWriteUserTo(User user, StreamOutput output) throws IOException {
+            if (user instanceof RunAsUser runAsUser) {
+                User.writeUser(user, output);
+                output.writeBoolean(true);
+                User.writeUser(runAsUser.authenticatingUser, output);
+            } else {
+                // no backcompat necessary, since there is no inner user
+                User.writeUser(user, output);
+            }
+            output.writeBoolean(false); // last user written, regardless of bwc, does not have an inner user
+        }
+    }
 }

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

@@ -18,7 +18,6 @@
 import org.elasticsearch.xpack.core.security.authc.Authentication;
 import org.elasticsearch.xpack.core.security.authz.accesscontrol.IndicesAccessControl;
 import org.elasticsearch.xpack.core.security.authz.privilege.ApplicationPrivilegeDescriptor;
-import org.elasticsearch.xpack.core.security.user.User;
 
 import java.util.Collection;
 import java.util.Collections;
@@ -53,8 +52,7 @@
  *         necessary to authorize the given user. It is important to note that the {@link RequestInfo}
  *         may contain an {@link Authentication} object that actually has two users when the
  *         <i>run as</i> feature is used and this method should resolve the information for both.
- *         To check for the presence of run as, use the {@link User#isRunAs()} method on the user
- *         retrieved using the {@link Authentication#getUser()} method.</li>
+ *         To check for the presence of run as, use the {@link Authentication#isRunAs()} method.</li>
  *     <li>{@link #authorizeRunAs(RequestInfo, AuthorizationInfo, ActionListener)} if the request
  *         is making use of the run as feature. This method is used to ensure the authenticated user
  *         can actually impersonate the user running the request.</li>

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

@@ -35,7 +35,6 @@ private AsyncSearchUser() {
         super(NAME, ROLE_NAME);
         // the following traits, and especially the run-as one, go with all the internal users
         // TODO abstract in a base `InternalUser` class
-        assert false == isRunAs() : "cannot run-as the system user";
         assert enabled();
         assert roles() != null && roles().length == 1;
     }