in_elasticsearch: respond to malformed bulk payloads

plugins/in_elasticsearch/in_elasticsearch_bulk_prot.c

@@ -256,10 +256,10 @@ static int process_ndpack(struct flb_in_elasticsearch *ctx, flb_sds_t tag, char
                     ret = flb_log_event_encoder_begin_record(encoder);
 
                     if (ret != FLB_EVENT_ENCODER_SUCCESS) {
-                        flb_sds_destroy(write_op);
                         flb_plg_error(ctx->ins, "event encoder error : %d", ret);
                         error_op = FLB_TRUE;
-
+                        ingest_result = -1;
+                        flb_sds_destroy(write_op);
                         break;
                     }
 
@@ -268,10 +268,10 @@ static int process_ndpack(struct flb_in_elasticsearch *ctx, flb_sds_t tag, char
                             &tm);
 
                     if (ret != FLB_EVENT_ENCODER_SUCCESS) {
-                        flb_sds_destroy(write_op);
                         flb_plg_error(ctx->ins, "event encoder error : %d", ret);
                         error_op = FLB_TRUE;
-
+                        ingest_result = -1;
+                        flb_sds_destroy(write_op);
                         break;
                     }
 
@@ -283,10 +283,10 @@ static int process_ndpack(struct flb_in_elasticsearch *ctx, flb_sds_t tag, char
                     }
 
                     if (ret != FLB_EVENT_ENCODER_SUCCESS) {
-                        flb_sds_destroy(write_op);
                         flb_plg_error(ctx->ins, "event encoder error : %d", ret);
                         error_op = FLB_TRUE;
-
+                        ingest_result = -1;
+                        flb_sds_destroy(write_op);
                         break;
                     }
                 }
@@ -310,7 +310,8 @@ static int process_ndpack(struct flb_in_elasticsearch *ctx, flb_sds_t tag, char
                     if (ret != FLB_EVENT_ENCODER_SUCCESS) {
                         flb_plg_error(ctx->ins, "event encoder error : %d", ret);
                         error_op = FLB_TRUE;
-
+                        ingest_result = -1;
+                        flb_sds_destroy(write_op);
                         break;
                     }
 
@@ -319,7 +320,8 @@ static int process_ndpack(struct flb_in_elasticsearch *ctx, flb_sds_t tag, char
                     if (ret != FLB_EVENT_ENCODER_SUCCESS) {
                         flb_plg_error(ctx->ins, "event encoder error : %d", ret);
                         error_op = FLB_TRUE;
-
+                        ingest_result = -1;
+                        flb_sds_destroy(write_op);
                         break;
                     }
 
@@ -418,14 +420,23 @@ static int process_ndpack(struct flb_in_elasticsearch *ctx, flb_sds_t tag, char
         else {
             flb_plg_error(ctx->ins, "skip record from invalid type: %i",
                          result.data.type);
+            flb_sds_destroy(write_op);
             msgpack_unpacked_destroy(&result);
             if (destroy_local_encoder == FLB_TRUE) {
                 flb_log_event_encoder_destroy(encoder);
             }
-            return -1;
+            return FLB_ERR_JSON_INVAL;
         }
     }
 
+    if (ingest_result != 0) {
+        msgpack_unpacked_destroy(&result);
+        if (destroy_local_encoder == FLB_TRUE) {
+            flb_log_event_encoder_destroy(encoder);
+        }
+        return ingest_result;
+    }
+
     if (idx % 2 != 0) {
         flb_plg_warn(ctx->ins, "decode payload of Bulk API is failed");
         msgpack_unpacked_destroy(&result);
@@ -440,19 +451,11 @@ static int process_ndpack(struct flb_in_elasticsearch *ctx, flb_sds_t tag, char
             flb_log_event_encoder_destroy(encoder);
         }
 
-        return -1;
+        return FLB_ERR_JSON_INVAL;
     }
 
     msgpack_unpacked_destroy(&result);
 
-    if (ingest_result != 0) {
-        if (destroy_local_encoder == FLB_TRUE) {
-            flb_log_event_encoder_destroy(encoder);
-        }
-
-        return ingest_result;
-    }
-
     if (destroy_local_encoder == FLB_TRUE) {
         flb_log_event_encoder_destroy(encoder);
     }
@@ -465,25 +468,40 @@ static ssize_t parse_payload_ndjson(struct flb_in_elasticsearch *ctx, flb_sds_t
 {
     int ret;
     int out_size;
+    size_t offset;
     char *pack;
     struct flb_pack_state pack_state;
 
     /* Initialize packer */
-    flb_pack_state_init(&pack_state);
+    ret = flb_pack_state_init(&pack_state);
+    if (ret != 0) {
+        return -1;
+    }
 
     /* Pack JSON as msgpack */
     ret = flb_pack_json_state(payload, size,
                               &pack, &out_size, &pack_state);
+    if (ret == 0) {
+        /* The packer can succeed with a complete prefix of an incomplete request. */
+        for (offset = pack_state.last_byte; offset < size; offset++) {
+            if (payload[offset] != ' ' && payload[offset] != '\t' &&
+                payload[offset] != '\r' && payload[offset] != '\n') {
+                flb_free(pack);
+                ret = FLB_ERR_JSON_INVAL;
+                break;
+            }
+        }
+    }
     flb_pack_state_reset(&pack_state);
 
     /* Handle exceptions */
     if (ret == FLB_ERR_JSON_PART) {
         flb_plg_warn(ctx->ins, "JSON data is incomplete, skipping");
-        return -1;
+        return FLB_ERR_JSON_INVAL;
     }
     else if (ret == FLB_ERR_JSON_INVAL) {
         flb_plg_warn(ctx->ins, "invalid JSON message, skipping");
-        return -1;
+        return FLB_ERR_JSON_INVAL;
     }
     else if (ret == -1) {
         return -1;
@@ -609,6 +627,8 @@ static int process_payload_ng(struct flb_http_request *request,
                               flb_sds_t tag,
                               flb_sds_t *bulk_statuses)
 {
+    int ret;
+
     if (request->content_type == NULL) {
         send_response_ng(response, 400, NULL, "error: header 'Content-Type' is not set\n");
 
@@ -627,8 +647,16 @@ static int process_payload_ng(struct flb_http_request *request,
         return -1;
     }
 
-    return parse_payload_ndjson(context, tag, request->body,
-                                cfl_sds_len(request->body), bulk_statuses);
+    ret = parse_payload_ndjson(context, tag, request->body,
+                               cfl_sds_len(request->body), bulk_statuses);
+    if (ret == FLB_ERR_JSON_INVAL) {
+        send_response_ng(response, 400, NULL, "error: invalid bulk payload\n");
+    }
+    else if (ret != 0 && ret != FLB_INPUT_INGRESS_BUSY) {
+        send_response_ng(response, 500, NULL, "error: could not ingest bulk payload\n");
+    }
+
+    return ret;
 }
 
 int in_elasticsearch_bulk_prot_handle_ng(struct flb_http_request *request,

tests/integration/scenarios/elasticsearch_invalid_bulk/tests/test_elasticsearch_invalid_bulk.py

@@ -0,0 +1,120 @@
+"""Reject malformed bulk payloads while keeping ingestion failures retryable."""
+import contextlib
+import http.client
+import json
+from pathlib import Path
+import socket
+import time
+
+import pytest
+
+from utils.fluent_bit_manager import FluentBitManager
+
+
+def wait_for(predicate, timeout=30):
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        if predicate():
+            return
+        time.sleep(0.05)
+    assert predicate(), "Timed out waiting for Fluent Bit output"
+
+
+@contextlib.contextmanager
+def daemon(tmp_path, mode, input_options=None):
+    with socket.socket() as listener:
+        listener.bind(("127.0.0.1", 0))
+        port = listener.getsockname()[1]
+    input_config = {"name": mode, "listen": "127.0.0.1", "port": port}
+    if input_options:
+        input_config.update(input_options)
+    config = {
+        "service": {
+            "flush": 0.1,
+            "grace": 1,
+            "http_server": "on",
+            "http_listen": "127.0.0.1",
+            "http_port": "${FLUENT_BIT_HTTP_MONITORING_PORT}",
+        },
+        "pipeline": {
+            "inputs": [input_config],
+            "outputs": [{"name": "stdout", "match": "*", "format": "json_lines"}],
+        },
+    }
+    config_path = tmp_path / "fluent-bit.yaml"
+    config_path.write_text(json.dumps(config))
+    manager = FluentBitManager(str(config_path))
+    try:
+        manager.start()
+        log = Path(manager.log_file)
+        yield ("127.0.0.1", port), manager.process, log
+    finally:
+        manager.stop()
+
+
+def send(mode, address, payload):
+    conn = http.client.HTTPConnection(*address, timeout=10)
+    try:
+        payload = b'{"index":{}}\n' + payload + b"\n"
+        conn.request("POST", "/_bulk", payload, {"Content-Type": "application/json"})
+        response = conn.getresponse()
+        response.read()
+        return response.status
+    finally:
+        conn.close()
+
+
+@pytest.mark.parametrize("mode", ["elasticsearch"])
+@pytest.mark.parametrize("kind", ["array", "map", "mixed"])
+def test_nested_json_recovery(tmp_path, mode, kind):
+    # Bounded regression data, with no process-crash or exploit-chain behavior.
+    value = b"0"
+    for index in range(65):
+        value = (b'{"k":' + value + b"}") if kind == "map" or (
+            kind == "mixed" and index % 2) else b"[" + value + b"]"
+    nested = b'{"nested":' + value + b"}"
+    with daemon(tmp_path, mode) as (address, process, log):
+        send(mode, address, b'{"marker":"before"}')
+        wait_for(lambda: '"marker":"before"' in log.read_text())
+        assert send(mode, address, nested) == 400
+        send(mode, address, b'{"marker":"after"}')
+        wait_for(lambda: '"marker":"after"' in log.read_text())
+        assert process.poll() is None
+        assert '"nested":' not in log.read_text()
+
+
+@pytest.mark.parametrize("payload", [b'{"broken":', b'not-json', b'[]', b'42'])
+def test_malformed_bulk_recovery(tmp_path, payload):
+    with daemon(tmp_path, "elasticsearch") as (address, process, log):
+        assert send("elasticsearch", address, payload) == 400
+        assert send("elasticsearch", address, b'{"marker":"after"}') == 200
+        wait_for(lambda: '"marker":"after"' in log.read_text())
+        assert process.poll() is None
+
+
+def test_busy_ingress_is_retryable(tmp_path):
+    options = {"http_server.workers": 2, "http_server.ingress_queue_byte_limit": "128"}
+    with daemon(tmp_path, "elasticsearch", options) as (address, process, log):
+        assert send("elasticsearch", address, json.dumps({"large": "x" * 1024}).encode()) == 503
+        assert send("elasticsearch", address, b'{"marker":"after"}') == 200
+        wait_for(lambda: '"marker":"after"' in log.read_text())
+        assert process.poll() is None
+
+
+@pytest.mark.parametrize("trailer", [b'{"index":', b'[', b'"unfinished', b'garbage'])
+def test_trailing_data_rejected_before_ingestion(tmp_path, trailer):
+    with daemon(tmp_path, "elasticsearch") as (address, process, log):
+        payload = b'{"marker":"rejected"}\n' + trailer
+        assert send("elasticsearch", address, payload) == 400
+        assert send("elasticsearch", address, b'{"marker":"after"}') == 200
+        wait_for(lambda: '"marker":"after"' in log.read_text())
+        assert '"marker":"rejected"' not in log.read_text()
+        assert process.poll() is None
+
+
+@pytest.mark.parametrize("trailer", [b'', b' \t\r\n \t'])
+def test_trailing_whitespace_accepted(tmp_path, trailer):
+    with daemon(tmp_path, "elasticsearch") as (address, process, log):
+        assert send("elasticsearch", address, b'{"marker":"accepted"}' + trailer) == 200
+        wait_for(lambda: '"marker":"accepted"' in log.read_text())
+        assert process.poll() is None

tests/runtime/in_elasticsearch.c

@@ -25,6 +25,7 @@
 #include <fluent-bit/flb_pack.h>
 #include <fluent-bit/flb_http_client.h>
 #include <fluent-bit/flb_gzip.h>
+#include <fluent-bit/flb_input.h>
 #include <monkey/mk_core.h>
 #include "flb_tests_runtime.h"
 
@@ -882,7 +883,51 @@ void flb_test_in_elasticsearch_index_op_with_plugin_tag()
     flb_test_in_elasticsearch("index", 9210, "es.index");
 }
 
+void flb_test_in_elasticsearch_ingestion_failure()
+{
+    struct flb_lib_out_cb cb_data = {0};
+    struct test_ctx *ctx;
+    struct flb_input_instance *input;
+    struct flb_http_client *client;
+    int ret;
+    size_t bytes_sent;
+    char *payload = "{\"index\":{}}\n{\"message\":\"valid\"}\n";
+
+    ctx = test_ctx_create(&cb_data);
+    if (!TEST_CHECK(ctx != NULL)) {
+        return;
+    }
+
+    ret = flb_input_set(ctx->flb, ctx->i_ffd, "port", "9211", NULL);
+    TEST_CHECK(ret == 0);
+    ret = flb_output_set(ctx->flb, ctx->o_ffd, "match", "*", NULL);
+    TEST_CHECK(ret == 0);
+    ret = flb_start(ctx->flb);
+    TEST_CHECK(ret == 0);
+
+    input = flb_input_get_instance(ctx->flb->config, ctx->i_ffd);
+    /* Force append failure without pausing the HTTP listener itself. */
+    input->mem_buf_status = FLB_INPUT_PAUSED;
+    ctx->httpc = in_elasticsearch_client_ctx_create(9211);
+    client = flb_http_client(ctx->httpc->u_conn, FLB_HTTP_POST, "/_bulk",
+                             payload, strlen(payload), "127.0.0.1", 9211, NULL, 0);
+    if (TEST_CHECK(client != NULL)) {
+        flb_http_add_header(client, FLB_HTTP_HEADER_CONTENT_TYPE,
+                           strlen(FLB_HTTP_HEADER_CONTENT_TYPE),
+                           NDJSON_CONTENT_TYPE, strlen(NDJSON_CONTENT_TYPE));
+        ret = flb_http_do(client, &bytes_sent);
+        TEST_CHECK(ret == 0);
+        TEST_CHECK(client->resp.status == 500);
+        TEST_MSG("expected HTTP 500, got %d", client->resp.status);
+        flb_http_client_destroy(client);
+    }
+    input->mem_buf_status = FLB_INPUT_RUNNING;
+    flb_upstream_conn_release(ctx->httpc->u_conn);
+    test_ctx_destroy(ctx);
+}
+
 TEST_LIST = {
+    {"ingestion_failure", flb_test_in_elasticsearch_ingestion_failure},
     {"version", flb_test_in_elasticsearch_version},
     {"configured_version", flb_test_in_elasticsearch_version_configured},
     {"index_op", flb_test_in_elasticsearch_index_op},