Handle streaming request body in audit log (#127843)

* Handle streaming request body in audit log

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.

Backport of #127798 to `8.17`

* 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";
+            }
             try {
                 return XContentHelper.convertToJson(request.content(), false, false, request.getXContentType());
             } catch (IOException ioe) {