Prevent ThreadContext header leak when sending response backport(#68649) (#85865)

We need to stash thread context in DefaultRestChannel before we call
channel.sendResponse because when calling this method, our execution
might be delayed and the thread be reused for another task - like
sending another response.
And it would see thread context from the initial “delayed” work.

This commit also expands the assertions on the empty thread context
to make sure it does not contains response headers.

closes #68278

Author: pgomulka

docs/changelog/68649.yaml

@@ -0,0 +1,6 @@
+pr: 68649
+summary: Prevent `ThreadContext` header leak when sending response
+area: Infra/Core
+type: bug
+issues:
+ - 68278

modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpRequestHandler.java

@@ -14,6 +14,7 @@
 
 import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.http.HttpPipelinedRequest;
+import org.elasticsearch.transport.Transports;
 
 @ChannelHandler.Sharable
 class Netty4HttpRequestHandler extends SimpleChannelInboundHandler<HttpPipelinedRequest> {
@@ -26,6 +27,8 @@ class Netty4HttpRequestHandler extends SimpleChannelInboundHandler<HttpPipelined
 
     @Override
     protected void channelRead0(ChannelHandlerContext ctx, HttpPipelinedRequest httpRequest) {
+        assert Transports.assertDefaultThreadContext(serverTransport.getThreadPool().getThreadContext());
+        assert Transports.assertTransportThread();
         final Netty4HttpChannel channel = ctx.channel().attr(Netty4HttpServerTransport.HTTP_CHANNEL_KEY).get();
         boolean success = false;
         try {
@@ -41,6 +44,8 @@ protected void channelRead0(ChannelHandlerContext ctx, HttpPipelinedRequest http
     @Override
     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
         ExceptionsHelper.maybeDieOnAnotherThread(cause);
+        assert Transports.assertDefaultThreadContext(serverTransport.getThreadPool().getThreadContext());
+
         Netty4HttpChannel channel = ctx.channel().attr(Netty4HttpServerTransport.HTTP_CHANNEL_KEY).get();
         if (cause instanceof Error) {
             serverTransport.onException(channel, new Exception(cause));

server/src/main/java/org/elasticsearch/http/AbstractHttpServerTransport.java

@@ -478,4 +478,8 @@ private static ActionListener<Void> earlyResponseListener(HttpRequest request, H
             return NO_OP;
         }
     }
+
+    public ThreadPool getThreadPool() {
+        return threadPool;
+    }
 }

server/src/main/java/org/elasticsearch/http/DefaultRestChannel.java

@@ -131,7 +131,9 @@ public void sendResponse(RestResponse restResponse) {
             addCookies(httpResponse);
 
             ActionListener<Void> listener = ActionListener.wrap(() -> Releasables.close(toClose));
-            httpChannel.sendResponse(httpResponse, listener);
+            try (ThreadContext.StoredContext existing = threadContext.stashContext()) {
+                httpChannel.sendResponse(httpResponse, listener);
+            }
             success = true;
         } finally {
             if (success == false) {