Drop project-id from threadcontext for CCS (elastic#136730)

When making a request to a remote cluster, we don't want to include
the current project-id ("X-Elastic-Project-Id") header in the thread
context that we send to the remote cluster because the current
project-id is not relevant to the thread context for the remote
execution.

Backport of: elastic#136664,elastic#136675

Author: tvernum

docs/changelog/136664.yaml

@@ -0,0 +1,5 @@
+pr: 136664
+summary: Drop project-id from threadcontext for CCS
+area: Authorization
+type: bug
+issues: []

server/src/main/java/org/elasticsearch/common/util/concurrent/ThreadContext.java

@@ -79,6 +79,11 @@ public final class ThreadContext implements Writeable, TraceContext {
     public static final String PREFIX = "request.headers";
     public static final Setting<Settings> DEFAULT_HEADERS_SETTING = Setting.groupSetting(PREFIX + ".", Property.NodeScope);
 
+    public enum HeadersFor {
+        LOCAL_CLUSTER,
+        REMOTE_CLUSTER
+    }
+
     /**
      * Name for the {@link #stashWithOrigin origin} attribute.
      */
@@ -108,14 +113,14 @@ public ThreadContext(Settings settings) {
      * @return a stored context that will restore the current context to its state at the point this method was called
      */
     public StoredContext stashContext() {
-        return stashContextPreservingRequestHeaders(Collections.emptySet());
+        return stashContextPreservingRequestHeaders(HeadersFor.LOCAL_CLUSTER, Collections.emptySet());
     }
 
     /**
      * Just like {@link #stashContext()} but also preserves request headers specified via {@code requestHeaders},
      * if these exist in the context before stashing.
      */
-    public StoredContext stashContextPreservingRequestHeaders(Set<String> requestHeaders) {
+    public StoredContext stashContextPreservingRequestHeaders(HeadersFor headersFor, Set<String> requestHeaders) {
         final ThreadContextStruct context = threadLocal.get();
 
         /*
@@ -126,7 +131,7 @@ public StoredContext stashContextPreservingRequestHeaders(Set<String> requestHea
          * This is needed so the DeprecationLogger in another thread can see the value of X-Opaque-ID provided by a user.
          * The same is applied to Task.TRACE_ID and other values specified in `Task.HEADERS_TO_COPY`.
          */
-        final Set<String> requestHeadersToCopy = getRequestHeadersToCopy(requestHeaders);
+        final Set<String> requestHeadersToCopy = getRequestHeadersToCopy(headersFor, requestHeaders);
         boolean hasHeadersToCopy = false;
         if (context.requestHeaders.isEmpty() == false) {
             for (String header : requestHeadersToCopy) {
@@ -157,8 +162,8 @@ public StoredContext stashContextPreservingRequestHeaders(Set<String> requestHea
         return storedOriginalContext(context);
     }
 
-    public StoredContext stashContextPreservingRequestHeaders(final String... requestHeaders) {
-        return stashContextPreservingRequestHeaders(Set.of(requestHeaders));
+    public StoredContext stashContextPreservingRequestHeaders(HeadersFor headersFor, final String... requestHeaders) {
+        return stashContextPreservingRequestHeaders(headersFor, Set.of(requestHeaders));
     }
 
     /**
@@ -275,12 +280,16 @@ private StoredContext storedOriginalContext(ThreadContextStruct originalContext)
         return () -> threadLocal.set(originalContext);
     }
 
-    private static Set<String> getRequestHeadersToCopy(Set<String> requestHeaders) {
-        if (requestHeaders.isEmpty()) {
+    private static Set<String> getRequestHeadersToCopy(HeadersFor headersFor, Set<String> requestHeaders) {
+        if (requestHeaders.isEmpty() && headersFor == HeadersFor.LOCAL_CLUSTER) {
             return HEADERS_TO_COPY;
         }
-        final Set<String> allRequestHeadersToCopy = new HashSet<>(requestHeaders);
-        allRequestHeadersToCopy.addAll(HEADERS_TO_COPY);
+        final Set<String> allRequestHeadersToCopy = new HashSet<>(HEADERS_TO_COPY);
+        if (headersFor == HeadersFor.REMOTE_CLUSTER) {
+            // Don't send the current project id to a remote cluster/project
+            allRequestHeadersToCopy.remove(Task.X_ELASTIC_PROJECT_ID_HTTP_HEADER);
+        }
+        allRequestHeadersToCopy.addAll(requestHeaders);
         return Set.copyOf(allRequestHeadersToCopy);
     }
 

server/src/test/java/org/elasticsearch/common/util/concurrent/ThreadContextTests.java

@@ -84,7 +84,14 @@ public void testStashContextPreservingRequestHeaders() {
         assertEquals("bar", threadContext.getHeader("foo"));
         assertEquals(Integer.valueOf(1), threadContext.getTransient("ctx.foo"));
         assertEquals("1", threadContext.getHeader("default"));
-        try (ThreadContext.StoredContext ignored = threadContext.stashContextPreservingRequestHeaders("foo", "ctx.foo", "missing")) {
+        try (
+            ThreadContext.StoredContext ignored = threadContext.stashContextPreservingRequestHeaders(
+                randomFrom(ThreadContext.HeadersFor.values()),
+                "foo",
+                "ctx.foo",
+                "missing"
+            )
+        ) {
             assertEquals("bar", threadContext.getHeader("foo"));
             // only request headers preserved, not transient
             assertNull(threadContext.getTransient("ctx.foo"));
@@ -99,15 +106,28 @@ public void testStashContextPreservingRequestHeaders() {
         assertEquals("1", threadContext.getHeader("default"));
     }
 
-    public void testStashContextPreservingHeadersWithDefaultHeadersToCopy() {
+    public void testStashContextPreservingHeadersWithDefaultHeadersToCopyInLocalCluster() {
+        doTestStashContextPreservingHeadersWithDefaultHeadersToCopy(ThreadContext.HeadersFor.LOCAL_CLUSTER);
+    }
+
+    public void testStashContextPreservingHeadersWithDefaultHeadersToCopyForRemoteCluster() {
+        doTestStashContextPreservingHeadersWithDefaultHeadersToCopy(ThreadContext.HeadersFor.REMOTE_CLUSTER);
+    }
+
+    private static void doTestStashContextPreservingHeadersWithDefaultHeadersToCopy(final ThreadContext.HeadersFor headersFor) {
         for (String header : HEADERS_TO_COPY) {
             ThreadContext threadContext = new ThreadContext(Settings.EMPTY);
             threadContext.putHeader(header, "bar");
-            try (ThreadContext.StoredContext ignored = threadContext.stashContextPreservingRequestHeaders()) {
-                assertEquals("bar", threadContext.getHeader(header));
+            try (ThreadContext.StoredContext ignored = threadContext.stashContextPreservingRequestHeaders(headersFor)) {
+                if (headersFor == ThreadContext.HeadersFor.REMOTE_CLUSTER && header.equals(Task.X_ELASTIC_PROJECT_ID_HTTP_HEADER)) {
+                    assertNull(threadContext.getHeader(header));
+                } else {
+                    assertEquals("bar", threadContext.getHeader(header));
+                }
             }
             // Also works if we pass it explicitly
-            try (ThreadContext.StoredContext ignored = threadContext.stashContextPreservingRequestHeaders(header)) {
+            try (ThreadContext.StoredContext ignored = threadContext.stashContextPreservingRequestHeaders(headersFor, header)) {
+                // If we pass it explicitly, then we expect it to be preserved even if it's not normally included in `HeadersFor`
                 assertEquals("bar", threadContext.getHeader(header));
             }
         }

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/transport/SecurityServerTransportInterceptor.java

@@ -426,7 +426,12 @@ private <T extends TransportResponse> void sendWithCrossClusterAccessHeaders(
             ) {
                 final ThreadContext threadContext = securityContext.getThreadContext();
                 final var contextRestoreHandler = new ContextRestoreResponseHandler<>(threadContext.newRestorableContext(true), handler);
-                try (ThreadContext.StoredContext ignored = threadContext.stashContextPreservingRequestHeaders(AuditUtil.AUDIT_REQUEST_ID)) {
+                try (
+                    ThreadContext.StoredContext ignored = threadContext.stashContextPreservingRequestHeaders(
+                        ThreadContext.HeadersFor.REMOTE_CLUSTER,
+                        AuditUtil.AUDIT_REQUEST_ID
+                    )
+                ) {
                     crossClusterAccessHeaders.writeToContext(threadContext);
                     sender.sendRequest(connection, action, request, options, contextRestoreHandler);
                 } catch (Exception e) {

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

@@ -728,6 +728,9 @@ private void doTestSendWithCrossClusterAccessHeaders(
     ) throws IOException {
         authentication.writeToContext(threadContext);
         final String expectedRequestId = AuditUtil.getOrGenerateRequestId(threadContext);
+        if (randomBoolean()) {
+            threadContext.putHeader(Task.X_ELASTIC_PROJECT_ID_HTTP_HEADER, randomProjectIdOrDefault().id());
+        }
         final String remoteClusterAlias = randomAlphaOfLengthBetween(5, 10);
         final String encodedApiKey = randomAlphaOfLengthBetween(10, 42);
         final String remoteClusterCredential = ApiKeyService.withApiKeyPrefix(encodedApiKey);
@@ -772,6 +775,7 @@ public <T extends TransportResponse> void sendRequest(
                 }
                 assertThat(securityContext.getAuthentication(), nullValue());
                 assertThat(AuditUtil.extractRequestId(securityContext.getThreadContext()), equalTo(expectedRequestId));
+                assertThat(threadContext.getHeader(Task.X_ELASTIC_PROJECT_ID_HTTP_HEADER), nullValue());
                 sentAction.set(action);
                 sentCredential.set(securityContext.getThreadContext().getHeader(CROSS_CLUSTER_ACCESS_CREDENTIALS_HEADER_KEY));
                 try {