Preserve request headers in a mixed version cluster (#79412)

When rewriting authentication for requests crossing nodes of different
versions, we now preserve all request headers except the authentication
one which needs to be rewritten. Previously all other request headers
were dropped and it caused issue like an operator user not being
recognised on the remote node. Other now preserved headers include audit
and system index access. This new behaviour is more correct because we
would never drop these headers if the nodes are on the same version.

Resolves: #79354

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

@@ -21,6 +21,7 @@
 import org.elasticsearch.node.Node;
 import org.elasticsearch.xpack.core.security.authc.Authentication;
 import org.elasticsearch.xpack.core.security.authc.Authentication.AuthenticationType;
+import org.elasticsearch.xpack.core.security.authc.AuthenticationField;
 import org.elasticsearch.xpack.core.security.authc.support.AuthenticationContextSerializer;
 import org.elasticsearch.xpack.core.security.authc.support.SecondaryAuthentication;
 import org.elasticsearch.xpack.core.security.user.User;
@@ -157,12 +158,19 @@ public <T> T executeWithAuthentication(Authentication authentication, Function<S
      * The original context is provided to the consumer. When this method returns, the original context is restored.
      */
     public void executeAfterRewritingAuthentication(Consumer<StoredContext> consumer, Version version) {
+        // Preserve request headers other than authentication
+        final Map<String, String> existingRequestHeaders = threadContext.getRequestHeadersOnly();
         final StoredContext original = threadContext.newStoredContext(true);
         final Authentication authentication = getAuthentication();
         try (ThreadContext.StoredContext ignore = threadContext.stashContext()) {
             setAuthentication(new Authentication(authentication.getUser(), authentication.getAuthenticatedBy(),
                 authentication.getLookedUpBy(), version, authentication.getAuthenticationType(),
                 rewriteMetadataForApiKeyRoleDescriptors(version, authentication)));
+            existingRequestHeaders.forEach((k, v) -> {
+                if (false == AuthenticationField.AUTHENTICATION_KEY.equals(k)) {
+                    threadContext.putHeader(k, v);
+                }
+            });
             consumer.accept(original);
         }
     }

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

@@ -33,6 +33,7 @@
 import static org.elasticsearch.xpack.core.security.authc.Authentication.VERSION_API_KEY_ROLES_AS_BYTES;
 import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY;
 import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.API_KEY_ROLE_DESCRIPTORS_KEY;
+import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.instanceOf;
 
 public class SecurityContextTests extends ESTestCase {
@@ -119,6 +120,11 @@ public void testExecuteAfterRewritingAuthentication() throws IOException {
         RealmRef authBy = new RealmRef("ldap", "foo", "node1");
         final Authentication original = new Authentication(user, authBy, authBy);
         original.writeToContext(threadContext);
+        final Map<String, String> requestHeaders = Map.of(
+            AuthenticationField.PRIVILEGE_CATEGORY_KEY, randomAlphaOfLengthBetween(3, 10),
+            randomAlphaOfLengthBetween(3, 8), randomAlphaOfLengthBetween(3, 8)
+        );
+        threadContext.putHeader(requestHeaders);
 
         final AtomicReference<StoredContext> contextAtomicReference = new AtomicReference<>();
         securityContext.executeAfterRewritingAuthentication(originalCtx -> {
@@ -129,6 +135,8 @@ public void testExecuteAfterRewritingAuthentication() throws IOException {
             assertEquals(VersionUtils.getPreviousVersion(), authentication.getVersion());
             assertEquals(original.getAuthenticationType(), securityContext.getAuthentication().getAuthenticationType());
             contextAtomicReference.set(originalCtx);
+            // Other request headers should be preserved
+            requestHeaders.forEach((k, v) -> assertThat(threadContext.getHeader(k), equalTo(v)));
         }, VersionUtils.getPreviousVersion());
 
         final Authentication authAfterExecution = securityContext.getAuthentication();

x-pack/qa/rolling-upgrade/build.gradle

@@ -66,6 +66,11 @@ BuildParams.bwcVersions.withWireCompatiple { bwcVersion, baseName ->
     if (bwcVersion.onOrAfter('6.6.0')) {
       setting 'ccr.auto_follow.wait_for_metadata_timeout', '1s'
     }
+    if (bwcVersion.onOrAfter('7.11.0')) {
+      extraConfigFile 'operator_users.yml', file("${project.projectDir}/src/test/resources/operator_users.yml")
+      setting 'xpack.security.operator_privileges.enabled', "true"
+      user username: "non_operator", password: 'x-pack-test-password', role: "superuser"
+    }
 
     user username: "test_user", password: "x-pack-test-password"
 
@@ -131,7 +136,8 @@ BuildParams.bwcVersions.withWireCompatiple { bwcVersion, baseName ->
       'mixed_cluster/90_ml_data_frame_analytics_crud/Put an outlier_detection job on the mixed cluster',
       'mixed_cluster/110_enrich/Enrich stats query smoke test for mixed cluster',
       'mixed_cluster/120_api_key/Test API key authentication will work in a mixed cluster',
-      'mixed_cluster/120_api_key/Create API key with metadata in a mixed cluster'
+      'mixed_cluster/120_api_key/Create API key with metadata in a mixed cluster',
+      'mixed_cluster/130_operator_privileges/Test operator privileges will work in the mixed cluster'
     ].join(',')
   }
 

x-pack/qa/rolling-upgrade/src/test/resources/operator_users.yml

@@ -0,0 +1,2 @@
+operator:
+  - usernames: ["test_user"]

x-pack/qa/rolling-upgrade/src/test/resources/rest-api-spec/test/mixed_cluster/130_operator_privileges.yml

@@ -0,0 +1,17 @@
+---
+"Test operator privileges will work in the mixed cluster":
+
+  - skip:
+      features: headers
+      version: " - 7.10.99"
+      reason: "operator privileges are available since 7.11"
+
+  # The default user ("test_user") is an operator, so this works
+  - do:
+      cluster.delete_voting_config_exclusions: { }
+
+  - do:
+      catch: forbidden
+      headers: # the non_operator user
+        Authorization: Basic bm9uX29wZXJhdG9yOngtcGFjay10ZXN0LXBhc3N3b3Jk
+      cluster.delete_voting_config_exclusions: { }