Add new audit handler for action responses (#63708)

This adds a new method to the AuditTrail that intercepts the
responses of transport-level actions. This new method is unlike all
the other existing audit methods because it is called after the
action has been run (so that it has access to the response).
After careful deliberation, the new method is called for the
responses of actions that are intercepted by the
`SecurityActionFilter` only, and not by the transport filter.

In order to facilitate the "linking" of the new audit event with the
other existing events, the audit method receives the requestId
as well as the authentication as arguments (in addition to the
request itself and the response).

This is labeled non-issue because it is only the foundation
upon which later features that actually print out (some) responses
can be built upon.

Related #63221

x-pack/plugin/ml/src/test/java/org/elasticsearch/xpack/ml/MlUpgradeModeActionFilterTests.java

@@ -97,7 +97,7 @@ public class MlUpgradeModeActionFilterTests extends ESTestCase {
 
     public void testOrder_UpgradeFilterIsExecutedAfterSecurityFilter() {
         MlUpgradeModeActionFilter upgradeModeFilter = new MlUpgradeModeActionFilter(clusterService);
-        SecurityActionFilter securityFilter = new SecurityActionFilter(null, null, null, mock(ThreadPool.class), null, null);
+        SecurityActionFilter securityFilter = new SecurityActionFilter(null, null, null, null, mock(ThreadPool.class), null, null);
 
         ActionFilter[] actionFiltersInOrderOfExecution = new ActionFilters(Sets.newHashSet(upgradeModeFilter, securityFilter)).filters();
         assertThat(actionFiltersInOrderOfExecution, is(arrayContaining(securityFilter, upgradeModeFilter)));

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

@@ -521,7 +521,7 @@ public class Security extends Plugin implements SystemIndexPlugin, IngestPlugin,
         securityInterceptor.set(new SecurityServerTransportInterceptor(settings, threadPool, authcService.get(),
                 authzService, getLicenseState(), getSslService(), securityContext.get(), destructiveOperations, clusterService));
 
-        securityActionFilter.set(new SecurityActionFilter(authcService.get(), authzService, getLicenseState(),
+        securityActionFilter.set(new SecurityActionFilter(authcService.get(), authzService, auditTrailService, getLicenseState(),
             threadPool, securityContext.get(), destructiveOperations));
 
         components.add(new SecurityUsageServices(realms, allRolesStore, nativeRoleMappingStore, ipFilter.get()));

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/action/filter/SecurityActionFilter.java

@@ -20,6 +20,7 @@ import org.elasticsearch.action.support.ActionFilter;
 import org.elasticsearch.action.support.ActionFilterChain;
 import org.elasticsearch.action.support.ContextPreservingActionListener;
 import org.elasticsearch.action.support.DestructiveOperations;
+import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.license.LicenseUtils;
 import org.elasticsearch.license.XPackLicenseState;
@@ -33,11 +34,12 @@ import org.elasticsearch.xpack.core.security.authz.privilege.HealthAndStatsPrivi
 import org.elasticsearch.xpack.core.security.support.Automatons;
 import org.elasticsearch.xpack.core.security.user.SystemUser;
 import org.elasticsearch.xpack.security.action.SecurityActionMapper;
+import org.elasticsearch.xpack.security.audit.AuditTrailService;
+import org.elasticsearch.xpack.security.audit.AuditUtil;
 import org.elasticsearch.xpack.security.authc.AuthenticationService;
 import org.elasticsearch.xpack.security.authz.AuthorizationService;
 import org.elasticsearch.xpack.security.authz.AuthorizationUtils;
 
-import java.io.IOException;
 import java.util.function.Predicate;
 
 public class SecurityActionFilter implements ActionFilter {
@@ -48,6 +50,7 @@ public class SecurityActionFilter implements ActionFilter {
 
     private final AuthenticationService authcService;
     private final AuthorizationService authzService;
+    private final AuditTrailService auditTrailService;
     private final SecurityActionMapper actionMapper = new SecurityActionMapper();
     private final XPackLicenseState licenseState;
     private final ThreadContext threadContext;
@@ -55,10 +58,11 @@ public class SecurityActionFilter implements ActionFilter {
     private final DestructiveOperations destructiveOperations;
 
     public SecurityActionFilter(AuthenticationService authcService, AuthorizationService authzService,
-                                XPackLicenseState licenseState, ThreadPool threadPool,
+                                AuditTrailService auditTrailService, XPackLicenseState licenseState, ThreadPool threadPool,
                                 SecurityContext securityContext, DestructiveOperations destructiveOperations) {
         this.authcService = authcService;
         this.authzService = authzService;
+        this.auditTrailService = auditTrailService;
         this.licenseState = licenseState;
         this.threadContext = threadPool.getThreadContext();
         this.securityContext = securityContext;
@@ -83,29 +87,19 @@ public class SecurityActionFilter implements ActionFilter {
         if (licenseState.isSecurityEnabled()) {
             final ActionListener<Response> contextPreservingListener =
                     ContextPreservingActionListener.wrapPreservingContext(listener, threadContext);
-            ActionListener<Void> authenticatedListener = ActionListener.wrap(
-                    (aVoid) -> chain.proceed(task, action, request, contextPreservingListener), contextPreservingListener::onFailure);
             final boolean useSystemUser = AuthorizationUtils.shouldReplaceUserWithSystem(threadContext, action);
             try {
                 if (useSystemUser) {
                     securityContext.executeAsUser(SystemUser.INSTANCE, (original) -> {
-                        try {
-                            applyInternal(action, request, authenticatedListener);
-                        } catch (IOException e) {
-                            listener.onFailure(e);
-                        }
+                        applyInternal(task, chain, action, request, contextPreservingListener);
                     }, Version.CURRENT);
                 } else if (AuthorizationUtils.shouldSetUserBasedOnActionOrigin(threadContext)) {
                     AuthorizationUtils.switchUserBasedOnActionOriginAndExecute(threadContext, securityContext, (original) -> {
-                        try {
-                            applyInternal(action, request, authenticatedListener);
-                        } catch (IOException e) {
-                            listener.onFailure(e);
-                        }
+                        applyInternal(task, chain, action, request, contextPreservingListener);
                     });
                 } else {
                     try (ThreadContext.StoredContext ignore = threadContext.newStoredContext(true)) {
-                        applyInternal(action, request, authenticatedListener);
+                        applyInternal(task, chain, action, request, contextPreservingListener);
                     }
                 }
             } catch (Exception e) {
@@ -130,13 +124,13 @@ public class SecurityActionFilter implements ActionFilter {
         return Integer.MIN_VALUE;
     }
 
-    private <Request extends ActionRequest> void applyInternal(String action, Request request,
-                                                               ActionListener<Void> listener) throws IOException {
+    private <Request extends ActionRequest, Response extends ActionResponse> void applyInternal(Task task,
+            ActionFilterChain<Request, Response> chain, String action, Request request, ActionListener<Response> listener) {
         if (CloseIndexAction.NAME.equals(action) || OpenIndexAction.NAME.equals(action) || DeleteIndexAction.NAME.equals(action)) {
             IndicesRequest indicesRequest = (IndicesRequest) request;
             try {
                 destructiveOperations.failDestructive(indicesRequest.indices());
-            } catch(IllegalArgumentException e) {
+            } catch (IllegalArgumentException e) {
                 listener.onFailure(e);
                 return;
             }
@@ -156,7 +150,17 @@ public class SecurityActionFilter implements ActionFilter {
         authcService.authenticate(securityAction, request, SystemUser.INSTANCE,
                 ActionListener.wrap((authc) -> {
                     if (authc != null) {
-                        authorizeRequest(authc, securityAction, request, listener);
+                        final String requestId = AuditUtil.extractRequestId(threadContext);
+                        assert Strings.hasText(requestId);
+                        authorizeRequest(authc, securityAction, request, ActionListener.delegateFailure(listener,
+                                (ignore, aVoid) -> {
+                                    chain.proceed(task, action, request, ActionListener.delegateFailure(listener,
+                                            (ignore2, response) -> {
+                                                auditTrailService.get().coordinatingActionResponse(requestId, authc, action, request,
+                                                        response);
+                                                listener.onResponse(response);
+                                            }));
+                                }));
                     } else if (licenseState.isSecurityEnabled() == false) {
                         listener.onResponse(null);
                     } else {
@@ -166,12 +170,11 @@ public class SecurityActionFilter implements ActionFilter {
     }
 
     private <Request extends ActionRequest> void authorizeRequest(Authentication authentication, String securityAction, Request request,
-                                                          ActionListener<Void> listener) {
+                                                                  ActionListener<Void> listener) {
         if (authentication == null) {
             listener.onFailure(new IllegalArgumentException("authentication must be non null for authorization"));
         } else {
-            authzService.authorize(authentication, securityAction, request, ActionListener.wrap(ignore -> listener.onResponse(null),
-                listener::onFailure));
+            authzService.authorize(authentication, securityAction, request, listener);
         }
     }
 }

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/audit/AuditTrail.java

@@ -8,6 +8,7 @@ package org.elasticsearch.xpack.security.audit;
 import org.elasticsearch.common.transport.TransportAddress;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.transport.TransportRequest;
+import org.elasticsearch.transport.TransportResponse;
 import org.elasticsearch.xpack.core.security.authc.Authentication;
 import org.elasticsearch.xpack.core.security.authc.AuthenticationToken;
 import org.elasticsearch.xpack.core.security.authz.AuthorizationEngine.AuthorizationInfo;
@@ -81,4 +82,9 @@ public interface AuditTrail {
     void explicitIndexAccessEvent(String requestId, AuditLevel eventType, Authentication authentication, String action, String indices,
                                   String requestName, TransportAddress remoteAddress, AuthorizationInfo authorizationInfo);
 
+    // this is the only audit method that is called *after* the action executed, when the response is available
+    // it is however *only called for coordinating actions*, which are the actions that a client invokes as opposed to
+    // the actions that a node invokes in order to service a client request
+    void coordinatingActionResponse(String requestId, Authentication authentication, String action, TransportRequest transportRequest,
+                                    TransportResponse transportResponse);
 }

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/audit/AuditTrailService.java

@@ -12,6 +12,7 @@ import org.elasticsearch.license.XPackLicenseState;
 import org.elasticsearch.license.XPackLicenseState.Feature;
 import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.transport.TransportRequest;
+import org.elasticsearch.transport.TransportResponse;
 import org.elasticsearch.xpack.core.security.authc.Authentication;
 import org.elasticsearch.xpack.core.security.authc.AuthenticationToken;
 import org.elasticsearch.xpack.core.security.authz.AuthorizationEngine.AuthorizationInfo;
@@ -147,6 +148,11 @@ public class AuditTrailService {
         public void explicitIndexAccessEvent(String requestId, AuditLevel eventType, Authentication authentication,
                                              String action, String indices, String requestName, TransportAddress remoteAddress,
                                              AuthorizationInfo authorizationInfo) {}
+
+        @Override
+        public void coordinatingActionResponse(String requestId, Authentication authentication, String action,
+                                               TransportRequest transportRequest,
+                                               TransportResponse transportResponse) { }
     }
 
     private static class CompositeAuditTrail implements AuditTrail {
@@ -254,6 +260,15 @@ public class AuditTrailService {
             }
         }
 
+        @Override
+        public void coordinatingActionResponse(String requestId, Authentication authentication, String action,
+                                               TransportRequest transportRequest,
+                                               TransportResponse transportResponse) {
+            for (AuditTrail auditTrail : auditTrails) {
+                auditTrail.coordinatingActionResponse(requestId, authentication, action, transportRequest, transportResponse);
+            }
+        }
+
         @Override
         public void tamperedRequest(String requestId, RestRequest request) {
             for (AuditTrail auditTrail : auditTrails) {

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/audit/logfile/LoggingAuditTrail.java

@@ -39,6 +39,7 @@ import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportRequest;
+import org.elasticsearch.transport.TransportResponse;
 import org.elasticsearch.xpack.core.security.action.CreateApiKeyAction;
 import org.elasticsearch.xpack.core.security.action.CreateApiKeyRequest;
 import org.elasticsearch.xpack.core.security.action.GrantApiKeyAction;
@@ -815,6 +816,13 @@ public class LoggingAuditTrail implements AuditTrail, ClusterStateListener {
         }
     }
 
+    @Override
+    public void coordinatingActionResponse(String requestId, Authentication authentication, String action,
+                                           TransportRequest transportRequest,
+                                           TransportResponse transportResponse) {
+        // not implemented yet
+    }
+
     private LogEntryBuilder securityChangeLogEntryBuilder(String requestId) {
         return new LogEntryBuilder(false)
                 .with(EVENT_TYPE_FIELD_NAME, SECURITY_CHANGE_ORIGIN_FIELD_VALUE)

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

@@ -61,11 +61,8 @@ import org.elasticsearch.xpack.core.security.authz.privilege.ApplicationPrivileg
 import org.elasticsearch.xpack.core.security.authz.privilege.ClusterPrivilegeResolver;
 import org.elasticsearch.xpack.core.security.authz.privilege.IndexPrivilege;
 import org.elasticsearch.xpack.core.security.user.AnonymousUser;
-import org.elasticsearch.xpack.core.security.user.AsyncSearchUser;
 import org.elasticsearch.xpack.core.security.user.SystemUser;
 import org.elasticsearch.xpack.core.security.user.User;
-import org.elasticsearch.xpack.core.security.user.XPackSecurityUser;
-import org.elasticsearch.xpack.core.security.user.XPackUser;
 import org.elasticsearch.xpack.security.audit.AuditLevel;
 import org.elasticsearch.xpack.security.audit.AuditTrail;
 import org.elasticsearch.xpack.security.audit.AuditTrailService;
@@ -96,6 +93,7 @@ import static org.elasticsearch.xpack.core.security.authz.AuthorizationServiceFi
 import static org.elasticsearch.xpack.core.security.authz.AuthorizationServiceField.INDICES_PERMISSIONS_KEY;
 import static org.elasticsearch.xpack.core.security.authz.AuthorizationServiceField.ORIGINATING_ACTION_KEY;
 import static org.elasticsearch.xpack.core.security.support.Exceptions.authorizationError;
+import static org.elasticsearch.xpack.core.security.user.User.isInternal;
 import static org.elasticsearch.xpack.security.audit.logfile.LoggingAuditTrail.PRINCIPAL_ROLES_FIELD_NAME;
 
 public class AuthorizationService {
@@ -191,7 +189,7 @@ public class AuthorizationService {
             if (auditId == null) {
                 // We would like to assert that there is an existing request-id, but if this is a system action, then that might not be
                 // true because the request-id is generated during authentication
-                if (isInternalUser(authentication.getUser()) != false) {
+                if (isInternal(authentication.getUser())) {
                     auditId = AuditUtil.getOrGenerateRequestId(threadContext);
                 } else {
                     auditTrailService.get().tamperedRequest(null, authentication, action, originalRequest);
@@ -199,6 +197,7 @@ public class AuthorizationService {
                             + "] without an existing request-id";
                     assert false : message;
                     listener.onFailure(new ElasticsearchSecurityException(message));
+                    return;
                 }
             }
 
@@ -398,7 +397,7 @@ public class AuthorizationService {
     private AuthorizationEngine getAuthorizationEngineForUser(final User user) {
         if (rbacEngine != authorizationEngine && licenseState.isSecurityEnabled() &&
             licenseState.checkFeature(Feature.SECURITY_AUTHORIZATION_ENGINE)) {
-            if (ClientReservedRealm.isReserved(user.principal(), settings) || isInternalUser(user)) {
+            if (ClientReservedRealm.isReserved(user.principal(), settings) || isInternal(user)) {
                 return rbacEngine;
             } else {
                 return authorizationEngine;
@@ -449,10 +448,6 @@ public class AuthorizationService {
         return request;
     }
 
-    private boolean isInternalUser(User user) {
-        return SystemUser.is(user) || XPackUser.is(user) || XPackSecurityUser.is(user) || AsyncSearchUser.is(user);
-    }
-
     private void authorizeRunAs(final RequestInfo requestInfo, final AuthorizationInfo authzInfo,
                                 final ActionListener<AuthorizationResult> listener) {
         final Authentication authentication = requestInfo.getAuthentication();

x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/action/filter/SecurityActionFilterTests.java

@@ -9,17 +9,18 @@ import org.apache.lucene.util.SetOnce;
 import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionRequest;
+import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.action.MockIndicesRequest;
 import org.elasticsearch.action.admin.indices.close.CloseIndexAction;
 import org.elasticsearch.action.admin.indices.delete.DeleteIndexAction;
 import org.elasticsearch.action.admin.indices.open.OpenIndexAction;
 import org.elasticsearch.action.support.ActionFilterChain;
-import org.elasticsearch.action.support.ContextPreservingActionListener;
 import org.elasticsearch.action.support.DestructiveOperations;
 import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.node.DiscoveryNodes;
+import org.elasticsearch.common.UUIDs;
 import org.elasticsearch.common.settings.ClusterSettings;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
@@ -37,6 +38,9 @@ import org.elasticsearch.xpack.core.security.authz.AuthorizationServiceField;
 import org.elasticsearch.xpack.core.security.authz.accesscontrol.IndicesAccessControl;
 import org.elasticsearch.xpack.core.security.user.SystemUser;
 import org.elasticsearch.xpack.core.security.user.User;
+import org.elasticsearch.xpack.security.audit.AuditTrail;
+import org.elasticsearch.xpack.security.audit.AuditTrailService;
+import org.elasticsearch.xpack.security.audit.AuditUtil;
 import org.elasticsearch.xpack.security.authc.AuthenticationService;
 import org.elasticsearch.xpack.security.authz.AuthorizationService;
 import org.junit.Before;
@@ -45,6 +49,7 @@ import java.util.Collections;
 
 import static org.elasticsearch.xpack.core.security.authz.AuthorizationServiceField.INDICES_PERMISSIONS_KEY;
 import static org.hamcrest.Matchers.arrayWithSize;
+import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.sameInstance;
 import static org.mockito.Matchers.any;
 import static org.mockito.Matchers.eq;
@@ -60,6 +65,9 @@ import static org.mockito.Mockito.when;
 public class SecurityActionFilterTests extends ESTestCase {
     private AuthenticationService authcService;
     private AuthorizationService authzService;
+    private AuditTrailService auditTrailService;
+    private AuditTrail auditTrail;
+    private ActionFilterChain chain;
     private XPackLicenseState licenseState;
     private SecurityActionFilter filter;
     private ThreadContext threadContext;
@@ -69,6 +77,10 @@ public class SecurityActionFilterTests extends ESTestCase {
     public void init() throws Exception {
         authcService = mock(AuthenticationService.class);
         authzService = mock(AuthorizationService.class);
+        auditTrailService = mock(AuditTrailService.class);
+        auditTrail = mock(AuditTrail.class);
+        when(auditTrailService.get()).thenReturn(auditTrail);
+        chain = mock(ActionFilterChain.class);
         licenseState = mock(XPackLicenseState.class);
         when(licenseState.isSecurityEnabled()).thenReturn(true);
         when(licenseState.checkFeature(Feature.SECURITY_STATS_AND_HEALTH)).thenReturn(true);
@@ -88,32 +100,37 @@ public class SecurityActionFilterTests extends ESTestCase {
         when(state.nodes()).thenReturn(nodes);
 
         SecurityContext securityContext = new SecurityContext(settings, threadContext);
-        filter = new SecurityActionFilter(authcService, authzService, licenseState, threadPool, securityContext, destructiveOperations);
+        filter = new SecurityActionFilter(authcService, authzService, auditTrailService, licenseState, threadPool,
+                securityContext, destructiveOperations);
     }
 
     public void testApply() throws Exception {
         ActionRequest request = mock(ActionRequest.class);
         ActionListener listener = mock(ActionListener.class);
-        ActionFilterChain chain = mock(ActionFilterChain.class);
         Task task = mock(Task.class);
         User user = new User("username", "r1", "r2");
         Authentication authentication = new Authentication(user, new RealmRef("test", "test", "foo"), null);
-        mockAuthentication(request, authentication);
+        String requestId = UUIDs.randomBase64UUID();
+        mockAuthentication(request, authentication, requestId);
         mockAuthorize();
+        ActionResponse actionResponse = mock(ActionResponse.class);
+        mockChain(task, "_action", request, actionResponse);
         filter.apply(task, "_action", request, listener, chain);
         verify(authzService).authorize(eq(authentication), eq("_action"), eq(request), any(ActionListener.class));
-        verify(chain).proceed(eq(task), eq("_action"), eq(request), isA(ContextPreservingActionListener.class));
+        verify(auditTrail).coordinatingActionResponse(eq(requestId), eq(authentication), eq("_action"), eq(request), eq(actionResponse));
     }
 
     public void testApplyRestoresThreadContext() throws Exception {
         ActionRequest request = mock(ActionRequest.class);
         ActionListener listener = mock(ActionListener.class);
-        ActionFilterChain chain = mock(ActionFilterChain.class);
         Task task = mock(Task.class);
         User user = new User("username", "r1", "r2");
         Authentication authentication = new Authentication(user, new RealmRef("test", "test", "foo"), null);
-        mockAuthentication(request, authentication);
+        String requestId = UUIDs.randomBase64UUID();
+        mockAuthentication(request, authentication, requestId);
         mockAuthorize();
+        ActionResponse actionResponse = mock(ActionResponse.class);
+        mockChain(task, "_action", request, actionResponse);
         assertNull(threadContext.getTransient(AuthenticationField.AUTHENTICATION_KEY));
         assertNull(threadContext.getTransient(INDICES_PERMISSIONS_KEY));
 
@@ -122,7 +139,7 @@ public class SecurityActionFilterTests extends ESTestCase {
         assertNull(threadContext.getTransient(AuthenticationField.AUTHENTICATION_KEY));
         assertNull(threadContext.getTransient(INDICES_PERMISSIONS_KEY));
         verify(authzService).authorize(eq(authentication), eq("_action"), eq(request), any(ActionListener.class));
-        verify(chain).proceed(eq(task), eq("_action"), eq(request), isA(ContextPreservingActionListener.class));
+        verify(auditTrail).coordinatingActionResponse(eq(requestId), eq(authentication), eq("_action"), eq(request), eq(actionResponse));
     }
 
     public void testApplyAsSystemUser() throws Exception {
@@ -132,15 +149,18 @@ public class SecurityActionFilterTests extends ESTestCase {
         Authentication authentication = new Authentication(user, new RealmRef("test", "test", "foo"), null);
         SetOnce<Authentication> authenticationSetOnce = new SetOnce<>();
         SetOnce<IndicesAccessControl> accessControlSetOnce = new SetOnce<>();
+        SetOnce<String> requestIdOnActionHandler = new SetOnce<>();
         ActionFilterChain chain = (task, action, request1, listener1) -> {
             authenticationSetOnce.set(threadContext.getTransient(AuthenticationField.AUTHENTICATION_KEY));
             accessControlSetOnce.set(threadContext.getTransient(INDICES_PERMISSIONS_KEY));
+            requestIdOnActionHandler.set(AuditUtil.extractRequestId(threadContext));
         };
         Task task = mock(Task.class);
         final boolean hasExistingAuthentication = randomBoolean();
         final boolean hasExistingAccessControl = randomBoolean();
         final String action = "internal:foo";
         if (hasExistingAuthentication) {
+            AuditUtil.generateRequestId(threadContext);
             threadContext.putTransient(AuthenticationField.AUTHENTICATION_KEY, authentication);
             threadContext.putHeader(AuthenticationField.AUTHENTICATION_KEY, "foo");
             threadContext.putTransient(AuthorizationServiceField.ORIGINATING_ACTION_KEY, "indices:foo");
@@ -148,12 +168,15 @@ public class SecurityActionFilterTests extends ESTestCase {
                 threadContext.putTransient(INDICES_PERMISSIONS_KEY, IndicesAccessControl.ALLOW_NO_INDICES);
             }
         } else {
+            assertNull(AuditUtil.extractRequestId(threadContext));
             assertNull(threadContext.getTransient(AuthenticationField.AUTHENTICATION_KEY));
         }
+        SetOnce<String> requestIdFromAuthn = new SetOnce<>();
         doAnswer(i -> {
             final Object[] args = i.getArguments();
             assertThat(args, arrayWithSize(4));
             ActionListener callback = (ActionListener) args[args.length - 1];
+            requestIdFromAuthn.set(AuditUtil.generateRequestId(threadContext));
             callback.onResponse(threadContext.getTransient(AuthenticationField.AUTHENTICATION_KEY));
             return Void.TYPE;
         }).when(authcService).authenticate(eq(action), eq(request), eq(SystemUser.INSTANCE), any(ActionListener.class));
@@ -174,6 +197,7 @@ public class SecurityActionFilterTests extends ESTestCase {
         assertNotEquals(authentication, authenticationSetOnce.get());
         assertEquals(SystemUser.INSTANCE, authenticationSetOnce.get().getUser());
         assertThat(accessControlSetOnce.get(), sameInstance(authzAccessControl));
+        assertThat(requestIdOnActionHandler.get(), is(requestIdFromAuthn.get()));
     }
 
     public void testApplyDestructiveOperations() throws Exception {
@@ -182,14 +206,19 @@ public class SecurityActionFilterTests extends ESTestCase {
                 randomFrom("*", "_all", "test*"));
         String action = randomFrom(CloseIndexAction.NAME, OpenIndexAction.NAME, DeleteIndexAction.NAME);
         ActionListener listener = mock(ActionListener.class);
-        ActionFilterChain chain = mock(ActionFilterChain.class);
         Task task = mock(Task.class);
         User user = new User("username", "r1", "r2");
         Authentication authentication = new Authentication(user, new RealmRef("test", "test", "foo"), null);
+        ActionResponse actionResponse = mock(ActionResponse.class);
+        mockChain(task, action, request, actionResponse);
+        SetOnce<String> requestIdFromAuthn = new SetOnce<>();
         doAnswer(i -> {
             final Object[] args = i.getArguments();
             assertThat(args, arrayWithSize(4));
             ActionListener callback = (ActionListener) args[args.length - 1];
+            requestIdFromAuthn.set(AuditUtil.generateRequestId(threadContext));
+            threadContext.putTransient(AuthenticationField.AUTHENTICATION_KEY, authentication);
+            threadContext.putHeader(AuthenticationField.AUTHENTICATION_KEY, authentication.encode());
             callback.onResponse(authentication);
             return Void.TYPE;
         }).when(authcService).authenticate(eq(action), eq(request), eq(SystemUser.INSTANCE), any(ActionListener.class));
@@ -202,10 +231,12 @@ public class SecurityActionFilterTests extends ESTestCase {
         filter.apply(task, action, request, listener, chain);
         if (failDestructiveOperations) {
             verify(listener).onFailure(isA(IllegalArgumentException.class));
-            verifyNoMoreInteractions(authzService, chain);
+            verifyNoMoreInteractions(authzService, chain, auditTrailService, auditTrail);
         } else {
             verify(authzService).authorize(eq(authentication), eq(action), eq(request), any(ActionListener.class));
-            verify(chain).proceed(eq(task), eq(action), eq(request), isA(ContextPreservingActionListener.class));
+            verify(chain).proceed(eq(task), eq(action), eq(request), any(ActionListener.class));
+            verify(auditTrail).coordinatingActionResponse(eq(requestIdFromAuthn.get()), eq(authentication), eq(action), eq(request),
+                    eq(actionResponse));
         }
     }
 
@@ -221,10 +252,21 @@ public class SecurityActionFilterTests extends ESTestCase {
             final Object[] args = i.getArguments();
             assertThat(args, arrayWithSize(4));
             ActionListener callback = (ActionListener) args[args.length - 1];
+            assertNull(threadContext.getTransient(AuthenticationField.AUTHENTICATION_KEY));
+            AuditUtil.generateRequestId(threadContext);
             callback.onResponse(authentication);
             return Void.TYPE;
         }).when(authcService).authenticate(eq("_action"), eq(request), eq(SystemUser.INSTANCE), any(ActionListener.class));
-        doThrow(exception).when(authzService).authorize(eq(authentication), eq("_action"), eq(request), any(ActionListener.class));
+        if (randomBoolean()) {
+            doThrow(exception).when(authzService).authorize(eq(authentication), eq("_action"), eq(request), any(ActionListener.class));
+        } else {
+            doAnswer((i) -> {
+                ActionListener<Void> callback = (ActionListener<Void>) i.getArguments()[3];
+                callback.onFailure(exception);
+                return Void.TYPE;
+            }).when(authzService)
+                    .authorize(eq(authentication), eq("_action"), eq(request), any(ActionListener.class));
+        }
         filter.apply(task, "_action", request, listener, chain);
         verify(listener).onFailure(exception);
         verifyNoMoreInteractions(chain);
@@ -242,13 +284,15 @@ public class SecurityActionFilterTests extends ESTestCase {
         verify(chain).proceed(eq(task), eq("_action"), eq(request), eq(listener));
     }
 
-    private void mockAuthentication(ActionRequest request, Authentication authentication) {
+    private void mockAuthentication(ActionRequest request, Authentication authentication, String requestId) {
         doAnswer(i -> {
             final Object[] args = i.getArguments();
             assertThat(args, arrayWithSize(4));
             ActionListener callback = (ActionListener) args[args.length - 1];
             assertNull(threadContext.getTransient(AuthenticationField.AUTHENTICATION_KEY));
             threadContext.putTransient(AuthenticationField.AUTHENTICATION_KEY, authentication);
+            threadContext.putHeader(AuthenticationField.AUTHENTICATION_KEY, authentication.encode());
+            threadContext.putHeader("_xpack_audit_request_id", requestId);
             callback.onResponse(authentication);
             return Void.TYPE;
         }).when(authcService).authenticate(eq("_action"), eq(request), eq(SystemUser.INSTANCE), any(ActionListener.class));
@@ -271,4 +315,13 @@ public class SecurityActionFilterTests extends ESTestCase {
                 .authorize(any(Authentication.class), any(String.class), any(TransportRequest.class), any(ActionListener.class));
     }
 
+    private void mockChain(Task task, String action, ActionRequest request, ActionResponse actionResponse) {
+        doAnswer(i -> {
+            final Object[] args = i.getArguments();
+            assertThat(args, arrayWithSize(4));
+            ActionListener callback = (ActionListener) args[args.length - 1];
+            callback.onResponse(actionResponse);
+            return Void.TYPE;
+        }).when(chain).proceed(eq(task), eq(action), eq(request), any(ActionListener.class));
+    }
 }