diff --git a/docs/changelog/130594.yaml b/docs/changelog/130594.yaml new file mode 100644 index 000000000000..41419cf48488 --- /dev/null +++ b/docs/changelog/130594.yaml @@ -0,0 +1,5 @@ +pr: 130594 +summary: Add audit logging for stream content +area: Network +type: enhancement +issues: [] diff --git a/x-pack/plugin/security/qa/audit/src/javaRestTest/java/org/elasticsearch/xpack/security/audit/AuditIT.java b/x-pack/plugin/security/qa/audit/src/javaRestTest/java/org/elasticsearch/xpack/security/audit/AuditIT.java index 9d6e49b63f39..fbc55948054e 100644 --- a/x-pack/plugin/security/qa/audit/src/javaRestTest/java/org/elasticsearch/xpack/security/audit/AuditIT.java +++ b/x-pack/plugin/security/qa/audit/src/javaRestTest/java/org/elasticsearch/xpack/security/audit/AuditIT.java @@ -109,10 +109,11 @@ public class AuditIT extends ESRestTestCase { public void testAuditAuthenticationSuccessForStreamingRequest() throws Exception { final Request request = new Request("POST", "/testindex/_bulk"); - request.setEntity(new StringEntity(""" + final String content = """ {"index":{}} {} - """, ContentType.create("application/x-ndjson", StandardCharsets.UTF_8))); + """; + request.setEntity(new StringEntity(content, ContentType.create("application/x-ndjson", StandardCharsets.UTF_8))); executeAndVerifyAudit( request, AuditLevel.AUTHENTICATION_SUCCESS, @@ -120,7 +121,7 @@ public class AuditIT extends ESRestTestCase { event, allOf( hasEntry(LoggingAuditTrail.AUTHENTICATION_TYPE_FIELD_NAME, "REALM"), - hasEntry(LoggingAuditTrail.REQUEST_BODY_FIELD_NAME, "Request body had not been received at the time of the audit event") + hasEntry(LoggingAuditTrail.REQUEST_BODY_FIELD_NAME, content) ) ) ); diff --git a/x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/audit/AuditTrailService.java b/x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/audit/AuditTrailService.java index 6fc70832cc38..46bbabcbb790 100644 --- a/x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/audit/AuditTrailService.java +++ b/x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/audit/AuditTrailService.java @@ -18,6 +18,7 @@ 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; import org.elasticsearch.xpack.security.Security; +import org.elasticsearch.xpack.security.audit.logfile.LoggingAuditTrail; import org.elasticsearch.xpack.security.transport.filter.SecurityIpFilterRule; import java.net.InetSocketAddress; @@ -53,6 +54,14 @@ public class AuditTrailService { } } + public boolean includeRequestBody() { + if (get() instanceof LoggingAuditTrail trail) { + return trail.includeRequestBody(); + } else { + return false; + } + } + // TODO: this method only exists for access to LoggingAuditTrail in a Node for testing. // DO NOT USE IT, IT WILL BE REMOVED IN THE FUTURE public AuditTrail getAuditTrail() { diff --git a/x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/audit/AuditUtil.java b/x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/audit/AuditUtil.java index fde2c2457d95..c584945bc3bd 100644 --- a/x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/audit/AuditUtil.java +++ b/x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/audit/AuditUtil.java @@ -27,9 +27,6 @@ public class AuditUtil { public static String restRequestContent(RestRequest request) { if (request.hasContent()) { - if (request.isStreamedContent()) { - return "Request body had not been received at the time of the audit event"; - } var content = request.content(); try { return XContentHelper.convertToJson(content, false, false, request.getXContentType()); diff --git a/x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/audit/logfile/LoggingAuditTrail.java b/x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/audit/logfile/LoggingAuditTrail.java index 4d56c44a1ca8..4de45cbc15c5 100644 --- a/x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/audit/logfile/LoggingAuditTrail.java +++ b/x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/audit/logfile/LoggingAuditTrail.java @@ -350,7 +350,7 @@ public class LoggingAuditTrail implements AuditTrail, ClusterStateListener { final EventFilterPolicyRegistry eventFilterPolicyRegistry; // package for testing volatile EnumSet events; - boolean includeRequestBody; + volatile boolean includeRequestBody; // fields that all entries have in common EntryCommonFields entryCommonFields; @@ -1072,6 +1072,10 @@ public class LoggingAuditTrail implements AuditTrail, ClusterStateListener { // not implemented yet } + public boolean includeRequestBody() { + return includeRequestBody; + } + private LogEntryBuilder securityChangeLogEntryBuilder(String requestId) { return new LogEntryBuilder(false).with(EVENT_TYPE_FIELD_NAME, SECURITY_CHANGE_ORIGIN_FIELD_VALUE).withRequestId(requestId); } diff --git a/x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/rest/SecurityRestFilter.java b/x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/rest/SecurityRestFilter.java index 6c3c25a95174..fa138193bbb0 100644 --- a/x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/rest/SecurityRestFilter.java +++ b/x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/rest/SecurityRestFilter.java @@ -22,7 +22,10 @@ import org.elasticsearch.xpack.security.authc.support.SecondaryAuthenticator; import org.elasticsearch.xpack.security.authz.restriction.WorkflowService; import org.elasticsearch.xpack.security.operator.OperatorPrivileges; +import java.util.function.Consumer; + import static org.elasticsearch.core.Strings.format; +import static org.elasticsearch.rest.RestContentAggregator.aggregate; public class SecurityRestFilter implements RestInterceptor { @@ -70,16 +73,29 @@ public class SecurityRestFilter implements RestInterceptor { return; } - final RestRequest wrappedRequest = maybeWrapRestRequest(request, targetHandler); - auditTrailService.get().authenticationSuccess(wrappedRequest); - secondaryAuthenticator.authenticateAndAttachToContext(wrappedRequest, ActionListener.wrap(secondaryAuthentication -> { - if (secondaryAuthentication != null) { - logger.trace("Found secondary authentication {} in REST request [{}]", secondaryAuthentication, request.uri()); - } - WorkflowService.resolveWorkflowAndStoreInThreadContext(targetHandler, threadContext); + // RestRequest might have stream content, in some cases we need to aggregate request content, for example audit logging. + final Consumer aggregationCallback = (aggregatedRestRequest) -> { + final RestRequest wrappedRequest = maybeWrapRestRequest(aggregatedRestRequest, targetHandler); + auditTrailService.get().authenticationSuccess(wrappedRequest); + secondaryAuthenticator.authenticateAndAttachToContext(wrappedRequest, ActionListener.wrap(secondaryAuthentication -> { + if (secondaryAuthentication != null) { + logger.trace( + "Found secondary authentication {} in REST request [{}]", + secondaryAuthentication, + aggregatedRestRequest.uri() + ); + } + WorkflowService.resolveWorkflowAndStoreInThreadContext(targetHandler, threadContext); + + doHandleRequest(aggregatedRestRequest, channel, targetHandler, listener); + }, e -> handleException(aggregatedRestRequest, e, listener))); + }; + if (request.isStreamedContent() && auditTrailService.includeRequestBody()) { + aggregate(request, aggregationCallback::accept); + } else { + aggregationCallback.accept(request); + } - doHandleRequest(request, channel, targetHandler, listener); - }, e -> handleException(request, e, listener))); } private void doHandleRequest(RestRequest request, RestChannel channel, RestHandler targetHandler, ActionListener listener) {