Support stored authentication headers prior to version 6.7 (#92221)

Officially Elasticsearch is compatible with last major at data level.
Therefore v8 is not compatible with v6. However we don't have a guided
migration path for stored authentication headers, e.g. upgrade assistant
does not do anything for them. Therefore it is more helpful and user
friendly for v8 to support v6 stored authentication headers.

This PR adds back the version conditional logic removed in #41185 along
with tests.

Author: ywangd

docs/changelog/92221.yaml

@@ -0,0 +1,5 @@
+pr: 92221
+summary: Support stored authentication headers prior to version 6.7
+area: Authentication
+type: bug
+issues: []

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

@@ -19,6 +19,7 @@
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.core.Nullable;
+import org.elasticsearch.core.Strings;
 import org.elasticsearch.xcontent.ConstructingObjectParser;
 import org.elasticsearch.xcontent.ParseField;
 import org.elasticsearch.xcontent.ToXContentObject;
@@ -90,6 +91,7 @@
 public final class Authentication implements ToXContentObject {
 
     private static final Logger logger = LogManager.getLogger(Authentication.class);
+    private static final Version VERSION_AUTHENTICATION_TYPE = Version.fromString("6.7.0");
 
     public static final Version VERSION_API_KEY_ROLES_AS_BYTES = Version.V_7_9_0;
     public static final Version VERSION_REALM_DOMAINS = Version.V_8_2_0;
@@ -141,8 +143,14 @@ public Authentication(StreamInput in) throws IOException {
         assert innerUser != null || lookedUpBy == null : "Authentication has no inner-user, but looked-up-by is [" + lookedUpBy + "]";
 
         final Version version = in.getVersion();
-        type = AuthenticationType.values()[in.readVInt()];
-        final Map<String, Object> metadata = in.readMap();
+        final Map<String, Object> metadata;
+        if (version.onOrAfter(VERSION_AUTHENTICATION_TYPE)) {
+            type = AuthenticationType.values()[in.readVInt()];
+            metadata = in.readMap();
+        } else {
+            type = AuthenticationType.REALM;
+            metadata = Map.of();
+        }
         if (innerUser != null) {
             authenticatingSubject = new Subject(innerUser, authenticatedBy, version, metadata);
             // The lookup user for run-as currently doesn't have authentication metadata associated with them because
@@ -469,8 +477,20 @@ public void writeTo(StreamOutput out) throws IOException {
         } else {
             out.writeBoolean(false);
         }
-        out.writeVInt(type.ordinal());
-        out.writeGenericMap(getAuthenticatingSubject().getMetadata());
+        final Map<String, Object> metadata = getAuthenticatingSubject().getMetadata();
+        if (out.getVersion().onOrAfter(VERSION_AUTHENTICATION_TYPE)) {
+            out.writeVInt(type.ordinal());
+            out.writeGenericMap(metadata);
+        } else {
+            assert type == AuthenticationType.REALM && metadata.isEmpty()
+                : Strings.format(
+                    "authentication with version [%s] must have authentication type %s and empty metadata, but got [%s] and [%s]",
+                    out.getVersion(),
+                    AuthenticationType.REALM,
+                    type,
+                    metadata
+                );
+        }
     }
 
     /**

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

@@ -22,6 +22,7 @@
 import org.elasticsearch.xpack.core.security.authc.esnative.NativeRealmSettings;
 import org.elasticsearch.xpack.core.security.authc.file.FileRealmSettings;
 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.User;
 
@@ -32,6 +33,7 @@
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
 
+import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasEntry;
 import static org.hamcrest.Matchers.hasKey;
 import static org.hamcrest.Matchers.is;
@@ -474,6 +476,34 @@ public void testToXContentWithServiceAccount() throws IOException {
         );
     }
 
+    public void testBwcWithStoredAuthenticationHeaders() throws IOException {
+        // Version 6.6.1
+        final String headerV6 = "p/HxAgANZWxhc3RpYy1hZG1pbgEJc3VwZXJ1c2VyCgAAAAEABG5vZGUFZmlsZTEEZmlsZQA=";
+        final Authentication authenticationV6 = AuthenticationContextSerializer.decode(headerV6);
+        assertThat(authenticationV6.getEffectiveSubject().getVersion(), equalTo(Version.fromString("6.6.1")));
+        assertThat(authenticationV6.getEffectiveSubject().getUser(), equalTo(new User("elastic-admin", "superuser")));
+        assertThat(authenticationV6.getAuthenticationType(), equalTo(Authentication.AuthenticationType.REALM));
+        assertThat(authenticationV6.isRunAs(), is(false));
+        assertThat(authenticationV6.encode(), equalTo(headerV6));
+
+        // Rewrite for a different version
+        final Version nodeVersion = VersionUtils.randomIndexCompatibleVersion(random());
+        final Authentication rewrittenAuthentication = authenticationV6.maybeRewriteForOlderVersion(nodeVersion);
+        assertThat(rewrittenAuthentication.getEffectiveSubject().getVersion(), equalTo(nodeVersion));
+        assertThat(rewrittenAuthentication.getEffectiveSubject().getUser(), equalTo(authenticationV6.getEffectiveSubject().getUser()));
+        assertThat(rewrittenAuthentication.getAuthenticationType(), equalTo(Authentication.AuthenticationType.REALM));
+        assertThat(rewrittenAuthentication.isRunAs(), is(false));
+
+        // Version 7.2.1
+        final String headerV7 = "p72sAwANZWxhc3RpYy1hZG1pbgENX2VzX3Rlc3Rfcm9vdAoAAAABAARub2RlBWZpbGUxBGZpbGUAAAoA";
+        final Authentication authenticationV7 = AuthenticationContextSerializer.decode(headerV7);
+        assertThat(authenticationV7.getEffectiveSubject().getVersion(), equalTo(Version.fromString("7.2.1")));
+        assertThat(authenticationV7.getEffectiveSubject().getUser(), equalTo(new User("elastic-admin", "_es_test_root")));
+        assertThat(authenticationV7.getAuthenticationType(), equalTo(Authentication.AuthenticationType.REALM));
+        assertThat(authenticationV7.isRunAs(), is(false));
+        assertThat(authenticationV7.encode(), equalTo(headerV7));
+    }
+
     private void runWithAuthenticationToXContent(Authentication authentication, Consumer<Map<String, Object>> consumer) throws IOException {
         try (XContentBuilder builder = XContentBuilder.builder(XContentType.JSON.xContent())) {
             authentication.toXContent(builder, ToXContent.EMPTY_PARAMS);