[PoC] Cross-project context preserving

plugins/examples/security-authorization-engine/src/main/java/org/elasticsearch/example/CustomAuthorizationEngine.java

@@ -87,14 +87,14 @@ public void authorizeClusterAction(RequestInfo requestInfo, AuthorizationInfo au
     }
 
     @Override
-    public SubscribableListener<IndexAuthorizationResult> authorizeIndexAction(
+    public void authorizeIndexAction(
         RequestInfo requestInfo,
         AuthorizationInfo authorizationInfo,
         AsyncSupplier<ResolvedIndices> indicesAsyncSupplier,
-        ProjectMetadata project
+        ProjectMetadata project,
+        ActionListener<IndexAuthorizationResult> listener
     ) {
         if (isSuperuser(requestInfo.getAuthentication().getEffectiveSubject().getUser())) {
-            SubscribableListener<IndexAuthorizationResult> listener = new SubscribableListener<>();
             indicesAsyncSupplier.getAsync().addListener(ActionListener.wrap(resolvedIndices -> {
                 Map<String, IndexAccessControl> indexAccessControlMap = new HashMap<>();
                 for (String name : resolvedIndices.getLocal()) {
@@ -104,9 +104,8 @@ public SubscribableListener<IndexAuthorizationResult> authorizeIndexAction(
                     new IndicesAccessControl(true, Collections.unmodifiableMap(indexAccessControlMap));
                 listener.onResponse(new IndexAuthorizationResult(indicesAccessControl));
             }, listener::onFailure));
-            return listener;
         } else {
-            return SubscribableListener.newSucceeded(new IndexAuthorizationResult(IndicesAccessControl.DENIED));
+            listener.onResponse(new IndexAuthorizationResult(IndicesAccessControl.DENIED));
         }
     }
 

server/src/main/java/org/elasticsearch/action/support/ContextPreservingActionListener.java

@@ -48,4 +48,16 @@ public void onFailure(Exception e) {
     public static <R> ContextPreservingActionListener<R> wrapPreservingContext(ActionListener<R> listener, ThreadContext threadContext) {
         return new ContextPreservingActionListener<>(threadContext.newRestorableContext(true), listener);
     }
+
+    public static <R> ContextPreservingActionListener<R> wrapPreservingTransientContext(
+        ActionListener<R> listener,
+        ThreadContext threadContext
+    ) {
+        return new ContextPreservingActionListener<>(
+            threadContext.wrapRestorable(
+                threadContext.newStoredContextPreservingResponseHeaders("_security_serverless_request_scoped_credential")
+            ),
+            listener
+        );
+    }
 }

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

@@ -364,6 +364,23 @@ public StoredContext newStoredContextPreservingResponseHeaders() {
         };
     }
 
+    public StoredContext newStoredContextPreservingResponseHeaders(String transientHeader) {
+        final ThreadContextStruct originalContext = threadLocal.get();
+        return () -> {
+            var found = threadLocal.get();
+            if (found != originalContext) {
+                if (found.transientHeaders.containsKey(transientHeader)) {
+                    threadLocal.set(
+                        originalContext.putResponseHeaders(found.responseHeaders)
+                            .putTransient(transientHeader, found.transientHeaders.get(transientHeader))
+                    );
+                } else {
+                    threadLocal.set(originalContext.putResponseHeaders(found.responseHeaders));
+                }
+            }
+        };
+    }
+
     /**
      * Capture the current context and then restore the given context, returning a {@link StoredContext} that reverts back to the current
      * context again. Equivalent to using {@link #newStoredContext()} and then calling {@code existingContext.restore()}.

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

@@ -12,7 +12,6 @@
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.IndicesRequest;
 import org.elasticsearch.action.support.IndexComponentSelector;
-import org.elasticsearch.action.support.SubscribableListener;
 import org.elasticsearch.cluster.metadata.IndexAbstraction;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.metadata.ProjectMetadata;
@@ -77,7 +76,7 @@
  *         can actually impersonate the user running the request.</li>
  *     <li>{@link #authorizeClusterAction(RequestInfo, AuthorizationInfo, ActionListener)} if the
  *         request is a cluster level operation.</li>
- *     <li>{@link #authorizeIndexAction(RequestInfo, AuthorizationInfo, AsyncSupplier, ProjectMetadata)} if
+ *     <li>{@link #authorizeIndexAction(RequestInfo, AuthorizationInfo, AsyncSupplier, ProjectMetadata,ActionListener)} if
  *         the request is a an index action. This method may be called multiple times for a single
  *         request as the request may be made up of sub-requests that also need to be authorized. The async supplier
  *         for resolved indices will invoke the
@@ -87,7 +86,7 @@
  * <br><p>
  * <em>NOTE:</em> the {@link #loadAuthorizedIndices(RequestInfo, AuthorizationInfo, Map, ActionListener)}
  * method may be called prior to
- * {@link #authorizeIndexAction(RequestInfo, AuthorizationInfo, AsyncSupplier, ProjectMetadata)}
+ * {@link #authorizeIndexAction(RequestInfo, AuthorizationInfo, AsyncSupplier, ProjectMetadata,ActionListener)}
  * in cases where wildcards need to be expanded.
  * </p><br>
  * Authorization engines can be called from various threads including network threads that should
@@ -163,13 +162,13 @@ public interface AuthorizationEngine {
      *                             attempting to operate on
      * @param metadata a map of a string name to the cluster metadata specific to that
      *                            alias or index
-     * @return a listener to be notified of the authorization result
      */
-    SubscribableListener<IndexAuthorizationResult> authorizeIndexAction(
+    void authorizeIndexAction(
         RequestInfo requestInfo,
         AuthorizationInfo authorizationInfo,
         AsyncSupplier<ResolvedIndices> indicesAsyncSupplier,
-        ProjectMetadata metadata
+        ProjectMetadata metadata,
+        ActionListener<IndexAuthorizationResult> listener
     );
 
     /**
@@ -779,6 +778,6 @@ interface AsyncSupplier<V> {
          * Asynchronously retrieves the value that is being supplied and notifies the listener upon
          * completion.
          */
-        SubscribableListener<V> getAsync();
+        void getAsync(ActionListener<V> listener);
     }
 }

x-pack/plugin/security/src/internalClusterTest/java/org/elasticsearch/xpack/security/profile/ProfileCancellationIntegTests.java

@@ -11,7 +11,6 @@
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.search.TransportSearchAction;
-import org.elasticsearch.action.support.SubscribableListener;
 import org.elasticsearch.client.Cancellable;
 import org.elasticsearch.client.Request;
 import org.elasticsearch.client.RequestOptions;
@@ -407,13 +406,14 @@ public void authorizeClusterAction(
                 }
 
                 @Override
-                public SubscribableListener<IndexAuthorizationResult> authorizeIndexAction(
+                public void authorizeIndexAction(
                     RequestInfo requestInfo,
                     AuthorizationInfo authorizationInfo,
                     AsyncSupplier<ResolvedIndices> indicesAsyncSupplier,
-                    ProjectMetadata metadata
+                    ProjectMetadata metadata,
+                    ActionListener<IndexAuthorizationResult> listener
                 ) {
-                    return SubscribableListener.newSucceeded(IndexAuthorizationResult.ALLOW_NO_INDICES);
+                    listener.onResponse(IndexAuthorizationResult.ALLOW_NO_INDICES);
                 }
 
                 @Override