Fix transport serialization of AsyncSearchUser (#54761)

This change ensures that the AsyncSearchUser is correctly (de)serialized when
an action executed by this user is sent to a remote node internally (via transport client).

x-pack/plugin/async-search/qa/security/build.gradle

@@ -10,6 +10,7 @@ dependencies {
 
 testClusters.integTest {
   testDistribution = 'DEFAULT'
+  numberOfNodes = 2
   setting 'xpack.license.self_generated.type', 'trial'
   setting 'xpack.security.enabled', 'true'
   extraConfigFile 'roles.yml', file('roles.yml')

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

@@ -21,6 +21,8 @@ public class InternalUserSerializationHelper {
                 return XPackUser.INSTANCE;
             } else if (XPackSecurityUser.is(username)) {
                 return XPackSecurityUser.INSTANCE;
+            } else if (AsyncSearchUser.is(username)) {
+                return AsyncSearchUser.INSTANCE;
             }
             throw new IllegalStateException("user [" + username + "] is not an internal user");
         }
@@ -36,6 +38,9 @@ public class InternalUserSerializationHelper {
         } else if (XPackSecurityUser.is(user)) {
             output.writeBoolean(true);
             output.writeString(XPackSecurityUser.NAME);
+        } else if (AsyncSearchUser.is(user)) {
+            output.writeBoolean(true);
+            output.writeString(AsyncSearchUser.NAME);
         } else {
             User.writeTo(user, output);
         }

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

@@ -59,6 +59,7 @@ import org.elasticsearch.xpack.core.security.authz.privilege.ApplicationPrivileg
 import org.elasticsearch.xpack.core.security.authz.privilege.ClusterPrivilegeResolver;
 import org.elasticsearch.xpack.core.security.authz.privilege.IndexPrivilege;
 import org.elasticsearch.xpack.core.security.user.AnonymousUser;
+import org.elasticsearch.xpack.core.security.user.AsyncSearchUser;
 import org.elasticsearch.xpack.core.security.user.SystemUser;
 import org.elasticsearch.xpack.core.security.user.User;
 import org.elasticsearch.xpack.core.security.user.XPackSecurityUser;
@@ -417,7 +418,7 @@ public class AuthorizationService {
     }
 
     private boolean isInternalUser(User user) {
-        return SystemUser.is(user) || XPackUser.is(user) || XPackSecurityUser.is(user);
+        return SystemUser.is(user) || XPackUser.is(user) || XPackSecurityUser.is(user) || AsyncSearchUser.is(user);
     }
 
     private void authorizeRunAs(final RequestInfo requestInfo, final AuthorizationInfo authzInfo,

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

@@ -7,6 +7,7 @@ package org.elasticsearch.xpack.security.user;
 
 import org.elasticsearch.common.io.stream.BytesStreamOutput;
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.xpack.core.security.user.AsyncSearchUser;
 import org.elasticsearch.xpack.core.security.user.ElasticUser;
 import org.elasticsearch.xpack.core.security.user.InternalUserSerializationHelper;
 import org.elasticsearch.xpack.core.security.user.KibanaUser;
@@ -87,6 +88,16 @@ public class UserSerializationTests extends ESTestCase {
         assertThat(readFrom.authenticatedUser(), is(XPackUser.INSTANCE));
     }
 
+    public void testAsyncSearchUserReadAndWrite() throws Exception {
+        BytesStreamOutput output = new BytesStreamOutput();
+
+        InternalUserSerializationHelper.writeTo(AsyncSearchUser.INSTANCE, output);
+        User readFrom = InternalUserSerializationHelper.readFrom(output.bytes().streamInput());
+
+        assertThat(readFrom, is(sameInstance(AsyncSearchUser.INSTANCE)));
+        assertThat(readFrom.authenticatedUser(), is(AsyncSearchUser.INSTANCE));
+    }
+
     public void testFakeInternalUserSerialization() throws Exception {
         BytesStreamOutput output = new BytesStreamOutput();
         output.writeBoolean(true);