[8.18] Handle streaming request body in audit log (elastic#127798) (elastic#127841)

* Handle streaming request body in audit log (elastic#127798)

The audit event for a successfully-authenticated REST request occurs
when we start to process the request. For APIs that accept a streaming
request body this means we have received the request headers, but not
its body, at the time of the audit event. Today such requests will fail
with a `ClassCastException` if the `emit_request_body` flag is set. This
change fixes the handling of streaming requests in the audit log to now
report that the request body was not available when writing the audit
entry.

* Enable incremental bulks in AuditIT

Author: DaveCTurner

docs/changelog/127798.yaml

@@ -0,0 +1,5 @@
+pr: 127798
+summary: Handle streaming request body in audit log
+area: Audit
+type: bug
+issues: []

x-pack/plugin/security/qa/audit/src/javaRestTest/java/org/elasticsearch/xpack/security/audit/AuditIT.java

@@ -7,6 +7,8 @@
 
 package org.elasticsearch.xpack.security.audit;
 
+import org.apache.http.entity.ContentType;
+import org.apache.http.entity.StringEntity;
 import org.elasticsearch.client.Request;
 import org.elasticsearch.client.RequestOptions;
 import org.elasticsearch.client.Response;
@@ -29,6 +31,7 @@
 import org.junit.ClassRule;
 
 import java.io.IOException;
+import java.nio.charset.StandardCharsets;
 import java.time.Instant;
 import java.time.ZonedDateTime;
 import java.time.format.DateTimeFormatter;
@@ -39,6 +42,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.function.Predicate;
 
+import static org.hamcrest.Matchers.allOf;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.hasEntry;
 import static org.hamcrest.Matchers.hasKey;
@@ -59,6 +63,7 @@ public class AuditIT extends ESRestTestCase {
         .setting("xpack.security.audit.enabled", "true")
         .setting("xpack.security.audit.logfile.events.include", "[ \"_all\" ]")
         .setting("xpack.security.audit.logfile.events.emit_request_body", "true")
+        .setting("rest.incremental_bulk", "true")
         .user("admin_user", "admin-password")
         .user(API_USER, "api-password", "superuser", false)
         .build();
@@ -105,6 +110,25 @@ public void testFilteringOfRequestBodies() throws Exception {
         });
     }
 
+    public void testAuditAuthenticationSuccessForStreamingRequest() throws Exception {
+        final Request request = new Request("POST", "/testindex/_bulk");
+        request.setEntity(new StringEntity("""
+            {"index":{}}
+            {}
+            """, ContentType.create("application/x-ndjson", StandardCharsets.UTF_8)));
+        executeAndVerifyAudit(
+            request,
+            AuditLevel.AUTHENTICATION_SUCCESS,
+            event -> assertThat(
+                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")
+                )
+            )
+        );
+    }
+
     private void executeAndVerifyAudit(Request request, AuditLevel eventType, CheckedConsumer<Map<String, Object>, Exception> assertions)
         throws Exception {
         Instant start = Instant.now();

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

@@ -27,6 +27,9 @@ 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());