Fix lingering license warning header in IP filter (#115510) (#115839)

Fixes another place where we do not stash thread context
that causes the license warning header to persist in
the thread context across Netty worker threads.

Resolves #114865

Relates to #107573

(cherry picked from commit fb9e028)

# Conflicts:
#	muted-tests.yml

docs/changelog/115510.yaml

@@ -0,0 +1,6 @@
+pr: 115510
+summary: Fix lingering license warning header in IP filter
+area: License
+type: bug
+issues:
+ - 114865

muted-tests.yml

@@ -330,8 +330,6 @@ tests:
 - class: org.elasticsearch.xpack.rank.rrf.RRFRankClientYamlTestSuiteIT
   method: test {yaml=rrf/800_rrf_with_text_similarity_reranker_retriever/explain using rrf retriever and text-similarity}
   issue: https://github.com/elastic/elasticsearch/issues/114757
-- class: org.elasticsearch.license.LicensingTests
-  issue: https://github.com/elastic/elasticsearch/issues/114865
 - class: org.elasticsearch.xpack.security.authc.ldap.GroupMappingIT
   issue: https://github.com/elastic/elasticsearch/issues/115221
 - class: org.elasticsearch.smoketest.DocsClientYamlTestSuiteIT
@@ -395,4 +393,4 @@ tests:
 #    issue: "https://github.com/elastic/elasticsearch/..."
 #  - class: "org.elasticsearch.xpack.esql.**"
 #    method: "test {union_types.MultiIndexIpStringStatsInline *}"
-#    issue: "https://github.com/elastic/elasticsearch/..."
+#    issue: "https://github.com/elastic/elasticsearch/..."

x-pack/plugin/security/src/internalClusterTest/java/org/elasticsearch/license/LicensingTests.java

@@ -41,6 +41,7 @@
 import org.elasticsearch.xpack.security.LocalStateSecurity;
 import org.hamcrest.Matchers;
 import org.junit.After;
+import org.junit.Assert;
 import org.junit.Before;
 
 import java.nio.file.Files;
@@ -241,6 +242,7 @@ public void testNoWarningHeaderWhenAuthenticationFailed() throws Exception {
         Header[] headers = null;
         try {
             getRestClient().performRequest(request);
+            Assert.fail("expected response exception");
         } catch (ResponseException e) {
             headers = e.getResponse().getHeaders();
             List<String> afterWarningHeaders = getWarningHeaders(headers);

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

@@ -10,6 +10,7 @@
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.ipfilter.AbstractRemoteAddressFilter;
 
+import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.xpack.security.transport.filter.IPFilter;
 
 import java.net.InetSocketAddress;
@@ -19,16 +20,21 @@ class IpFilterRemoteAddressFilter extends AbstractRemoteAddressFilter<InetSocket
 
     private final IPFilter filter;
     private final String profile;
+    private final ThreadContext threadContext;
 
-    IpFilterRemoteAddressFilter(final IPFilter filter, final String profile) {
+    IpFilterRemoteAddressFilter(final IPFilter filter, final String profile, final ThreadContext threadContext) {
         this.filter = filter;
         this.profile = profile;
+        this.threadContext = threadContext;
     }
 
     @Override
     protected boolean accept(final ChannelHandlerContext ctx, final InetSocketAddress remoteAddress) throws Exception {
         // at this stage no auth has happened, so we do not have any principal anyway
-        return filter.accept(profile, remoteAddress);
+        // this prevents thread-context changes to propagate beyond the channel accept test, as netty worker threads are reused
+        try (ThreadContext.StoredContext ignore = threadContext.newStoredContext()) {
+            return filter.accept(profile, remoteAddress);
+        }
     }
 
 }

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

@@ -104,7 +104,7 @@ protected void initChannel(final Channel ch) throws Exception {
 
     private void maybeAddIPFilter(final Channel ch, final String name) {
         if (authenticator != null) {
-            ch.pipeline().addFirst("ipfilter", new IpFilterRemoteAddressFilter(authenticator, name));
+            ch.pipeline().addFirst("ipfilter", new IpFilterRemoteAddressFilter(authenticator, name, getThreadPool().getThreadContext()));
         }
     }
 

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

@@ -15,6 +15,7 @@
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.transport.BoundTransportAddress;
 import org.elasticsearch.common.transport.TransportAddress;
+import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.http.HttpServerTransport;
 import org.elasticsearch.license.MockLicenseState;
 import org.elasticsearch.license.TestUtils;
@@ -90,10 +91,11 @@ public void init() throws Exception {
             ipFilter.setBoundHttpTransportAddress(httpTransport.boundAddress());
         }
 
+        ThreadContext threadContext = new ThreadContext(Settings.EMPTY);
         if (isHttpEnabled) {
-            handler = new IpFilterRemoteAddressFilter(ipFilter, IPFilter.HTTP_PROFILE_NAME);
+            handler = new IpFilterRemoteAddressFilter(ipFilter, IPFilter.HTTP_PROFILE_NAME, threadContext);
         } else {
-            handler = new IpFilterRemoteAddressFilter(ipFilter, "default");
+            handler = new IpFilterRemoteAddressFilter(ipFilter, "default", threadContext);
         }
     }
 
@@ -106,7 +108,11 @@ public void testThatFilteringWorksByIp() throws Exception {
     }
 
     public void testFilteringWorksForRemoteClusterPort() throws Exception {
-        handler = new IpFilterRemoteAddressFilter(ipFilter, RemoteClusterPortSettings.REMOTE_CLUSTER_PROFILE);
+        handler = new IpFilterRemoteAddressFilter(
+            ipFilter,
+            RemoteClusterPortSettings.REMOTE_CLUSTER_PROFILE,
+            new ThreadContext(Settings.EMPTY)
+        );
         InetSocketAddress localhostAddr = new InetSocketAddress(InetAddresses.forString("127.0.0.1"), 12345);
         assertThat(handler.accept(mock(ChannelHandlerContext.class), localhostAddr), is(true));