pack: bound JSON conversion depth and handle packing failures

src/flb_pack.c

@@ -314,61 +314,65 @@ static inline int pack_string_token(struct flb_pack_state *state,
 
 /* Convert a yyjson value to msgpack */
 #ifdef FLB_HAVE_YYJSON
-static void yyjson_val_to_msgpack(yyjson_val *val, msgpack_packer *pck)
+static int yyjson_val_to_msgpack(yyjson_val *val, msgpack_packer *pck, size_t depth)
 {
-    size_t idx, max;
+    size_t idx;
+    size_t max;
     yyjson_val *key;
     yyjson_val *tmp;
     const char *k;
     size_t klen;
 
+    /* Bound recursion before entering a container, as the unpacker does. */
+    if (yyjson_is_ctn(val) && depth >= MSGPACK_EMBED_STACK_SIZE) {
+        return -1;
+    }
+
     switch (yyjson_get_type(val)) {
     case YYJSON_TYPE_OBJ:
-        msgpack_pack_map(pck, yyjson_obj_size(val));
+        if (msgpack_pack_map(pck, yyjson_obj_size(val)) != 0) {
+            return -1;
+        }
         yyjson_obj_foreach(val, idx, max, key, tmp) {
             k = yyjson_get_str(key);
             klen = yyjson_get_len(key);
-            msgpack_pack_str(pck, klen);
-            msgpack_pack_str_body(pck, k, klen);
-            yyjson_val_to_msgpack(tmp, pck);
+            if (msgpack_pack_str(pck, klen) != 0 ||
+                msgpack_pack_str_body(pck, k, klen) != 0 ||
+                yyjson_val_to_msgpack(tmp, pck, depth + 1) != 0) {
+                return -1;
+            }
         }
-        break;
+        return 0;
     case YYJSON_TYPE_ARR:
-        msgpack_pack_array(pck, yyjson_arr_size(val));
+        if (msgpack_pack_array(pck, yyjson_arr_size(val)) != 0) {
+            return -1;
+        }
         yyjson_arr_foreach(val, idx, max, tmp) {
-            yyjson_val_to_msgpack(tmp, pck);
+            if (yyjson_val_to_msgpack(tmp, pck, depth + 1) != 0) {
+                return -1;
+            }
         }
-        break;
+        return 0;
     case YYJSON_TYPE_STR:
-        msgpack_pack_str(pck, yyjson_get_len(val));
-        msgpack_pack_str_body(pck, yyjson_get_str(val), yyjson_get_len(val));
-        break;
+        if (msgpack_pack_str(pck, yyjson_get_len(val)) != 0) {
+            return -1;
+        }
+        return msgpack_pack_str_body(pck, yyjson_get_str(val), yyjson_get_len(val));
     case YYJSON_TYPE_BOOL:
         if (yyjson_get_bool(val)) {
-            msgpack_pack_true(pck);
-        }
-        else {
-            msgpack_pack_false(pck);
+            return msgpack_pack_true(pck);
         }
-        break;
-    case YYJSON_TYPE_NULL:
-        msgpack_pack_nil(pck);
-        break;
+        return msgpack_pack_false(pck);
     case YYJSON_TYPE_NUM:
         if (yyjson_is_int(val)) {
             if (yyjson_is_sint(val)) {
-                msgpack_pack_int64(pck, yyjson_get_sint(val));
-            }
-            else {
-                msgpack_pack_uint64(pck, yyjson_get_uint(val));
+                return msgpack_pack_int64(pck, yyjson_get_sint(val));
             }
+            return msgpack_pack_uint64(pck, yyjson_get_uint(val));
         }
-        else {
-            msgpack_pack_double(pck, yyjson_get_real(val));
-        }
-        break;
+        return msgpack_pack_double(pck, yyjson_get_real(val));
     default:
-        msgpack_pack_nil(pck);
+        return msgpack_pack_nil(pck);
     }
 }
 
@@ -399,7 +403,7 @@ static int pack_json_to_msgpack_yyjson(const char *js, size_t len, char **buffer
     msgpack_packer pck;
     char *start, *end, *insitu_buf;
 
-    if (!js || !buffer || !size) {
+    if (!js || !buffer || !size || len > SIZE_MAX - YYJSON_PADDING_SIZE) {
         return -1;
     }
 
@@ -478,7 +482,12 @@ static int pack_json_to_msgpack_yyjson(const char *js, size_t len, char **buffer
             return -1;
         }
 
-        yyjson_val_to_msgpack(root, &pck);
+        if (yyjson_val_to_msgpack(root, &pck, 0) != 0) {
+            yyjson_doc_free(doc);
+            msgpack_sbuffer_destroy(&sbuf);
+            flb_free(insitu_buf);
+            return -1;
+        }
 
         if (root_type && count_records == 0) {
             *root_type = yyjson_root_type(root);

tests/integration/scenarios/security_ingestors/tests/test_security_ingestors.py

@@ -0,0 +1,154 @@
+"""Bounded nesting and recovery checks for network ingestion."""
+import contextlib
+import http.client
+import os
+from pathlib import Path
+import signal
+import socket
+import struct
+import subprocess
+import time
+
+import pytest
+
+
+
+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"
+
+
+@contextlib.contextmanager
+def daemon(tmp_path, mode):
+    with socket.socket() as listener:
+        listener.bind(("127.0.0.1", 0))
+        port = listener.getsockname()[1]
+    plugin = mode.split("-")[0]
+    address = str(tmp_path / "input.sock") if plugin == "unix_socket" else ("127.0.0.1", port)
+    parser = tmp_path / "parsers.conf"
+    parser.write_text("[PARSER]\n    Name json\n    Format json\n")
+    command = [os.environ["FLUENT_BIT_BINARY"], "-f", "0.1", "-R", str(parser), "-i", plugin]
+    if plugin == "unix_socket":
+        command += ["-p", f"socket_path={address}"]
+    else:
+        command += ["-p", "listen=127.0.0.1", "-p", f"port={port}"]
+    if mode.endswith("-parser"):
+        command += ["-p", "format=none", "-p", "parser=json"]
+    if plugin == "syslog":
+        command += ["-p", "mode=tcp", "-p", "parser=json"]
+    command += ["-o", "stdout", "-m", "*", "-p", "format=json_lines"]
+    log = tmp_path / "fluent-bit.log"
+    memlog = tmp_path / "valgrind.log"
+    memory = os.environ.get("VALGRIND") == "1"
+    if memory:
+        command = ["valgrind", "--leak-check=full", "--show-leak-kinds=all",
+                   "--errors-for-leak-kinds=definite,indirect", "--error-exitcode=99",
+                   f"--log-file={memlog}"] + command
+    with log.open("w") as output:
+        process = subprocess.Popen(command, stdout=output, stderr=subprocess.STDOUT)
+        def ready():
+            assert process.poll() is None, log.read_text()
+            return "[output:stdout:" in log.read_text()
+        try:
+            wait_for(ready)
+            yield address, process, log
+        finally:
+            if process.poll() is None:
+                process.send_signal(signal.SIGTERM)
+            try:
+                process.wait(timeout=30)
+            except subprocess.TimeoutExpired:
+                process.kill()
+                process.wait()
+                pytest.fail("Fluent Bit did not shut down cleanly")
+            assert process.returncode == 0, log.read_text() + (memlog.read_text() if memory else "")
+            if memory:
+                assert "ERROR SUMMARY: 0 errors" in memlog.read_text(), memlog.read_text()
+
+
+def send(mode, address, payload):
+    plugin = mode.split("-")[0]
+    if plugin in ("http", "splunk", "elasticsearch"):
+        conn = http.client.HTTPConnection(*address, timeout=10)
+        try:
+            path = "/test"
+            if plugin == "splunk":
+                path = "/services/collector/event"
+                payload = b'{"event":' + payload + b"}"
+            elif plugin == "elasticsearch":
+                path = "/_bulk"
+                payload = b'{"index":{}}\n' + payload + b"\n"
+            conn.request("POST", path, payload, {"Content-Type": "application/json"})
+            response = conn.getresponse()
+            response.read()
+        finally:
+            conn.close()
+        return
+    if plugin == "udp":
+        with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
+            sock.sendto(payload + b"\n", address)
+        return
+    family = socket.AF_UNIX if plugin == "unix_socket" else socket.AF_INET
+    with socket.socket(family, socket.SOCK_STREAM) as sock:
+        sock.settimeout(5)
+        sock.connect(address)
+        if plugin == "mqtt":
+            # A regular MQTT CONNECT followed by a QoS 0 JSON publication.
+            sock.sendall(b"\x10\x10\x00\x04MQTT\x04\x02\x00\x0a\x00\x04test")
+            assert sock.recv(4)[0] == 0x20
+            body = b"\x00\x01a" + payload
+            length = len(body)
+            encoded = bytearray()
+            while True:
+                digit = length % 128
+                length //= 128
+                encoded.append(digit | (0x80 if length else 0))
+                if not length:
+                    break
+            sock.sendall(b"\x30" + bytes(encoded) + body)
+        else:
+            sock.sendall(payload if plugin == "forward" else payload + b"\n")
+
+
+@pytest.mark.parametrize("mode", ["mqtt", "tcp-parser", "udp-parser", "tcp", "udp", "http",
+                                 "splunk"])
+@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())
+        send(mode, address, nested)
+        send(mode, address, b'{"marker":"after"}')
+        wait_for(lambda: '"marker":"after"' in log.read_text())
+        assert process.poll() is None
+        # Parser inputs may preserve rejected JSON as a raw log string.
+        assert '"nested":' not in log.read_text()
+
+
+@pytest.mark.parametrize("kind", ["array", "map"])
+def test_forward_log_depth_recovery(tmp_path, kind):
+    def message(record):
+        return b"\x93\xa4test\x01" + record
+    def marker(value):
+        return b"\x81\xa6marker" + bytes([0xa0 | len(value)]) + value
+    nested = b"\x00"
+    for _ in range(65):
+        nested = (b"\x81\xa1k" if kind == "map" else b"\x91") + nested
+    with daemon(tmp_path, "forward") as (address, process, log):
+        send("forward", address, message(marker(b"before")))
+        wait_for(lambda: '"marker":"before"' in log.read_text())
+        send("forward", address, message(b"\x81\xa6nested" + nested))
+        send("forward", address, message(marker(b"after")))
+        wait_for(lambda: '"marker":"after"' in log.read_text())
+        assert process.poll() is None
+        assert '"nested":' not in log.read_text()

tests/internal/pack.c

@@ -1287,7 +1287,75 @@ void test_json_pack_token_count_overflow()
     flb_pack_state_reset(&state);
 }
 
+#ifdef FLB_HAVE_YYJSON
+static void test_json_pack_depth_limit(void)
+{
+    int kind;
+    int depth;
+    int index;
+    int ret;
+    int root_type;
+    int records;
+    int prefix;
+    size_t offset;
+    size_t size;
+    char *buffer;
+    flb_sds_t json;
+    msgpack_unpacked unpacked;
+
+    /* Exercise maps, arrays, mixed nesting, and cleanup after a valid record. */
+    for (kind = 0; kind < 3; kind++) {
+        for (prefix = 0; prefix < 2; prefix++) {
+            for (depth = MSGPACK_EMBED_STACK_SIZE - 1;
+                 depth <= MSGPACK_EMBED_STACK_SIZE + 1; depth++) {
+                json = flb_sds_create(prefix ? "{} " : "");
+                TEST_ASSERT(json != NULL);
+                for (index = 0; index < depth; index++) {
+                    json = flb_sds_cat(json, kind == 0 || (kind == 2 && index % 2) ?
+                                      "{\"k\":" : "[",
+                                      kind == 0 || (kind == 2 && index % 2) ? 5 : 1);
+                    TEST_ASSERT(json != NULL);
+                }
+                json = flb_sds_cat(json, "0", 1);
+                TEST_ASSERT(json != NULL);
+                for (index = depth - 1; index >= 0; index--) {
+                    json = flb_sds_cat(json, kind == 0 || (kind == 2 && index % 2) ?
+                                      "}" : "]", 1);
+                    TEST_ASSERT(json != NULL);
+                }
+                buffer = NULL;
+                size = 0;
+                ret = flb_pack_json_recs(json, flb_sds_len(json), &buffer, &size,
+                                         &root_type, &records, NULL);
+                if (depth > MSGPACK_EMBED_STACK_SIZE) {
+                    TEST_CHECK(ret != 0);
+                    TEST_CHECK(buffer == NULL);
+                    TEST_CHECK(size == 0);
+                }
+                else {
+                    TEST_ASSERT(ret == 0);
+                    TEST_CHECK(records == prefix + 1);
+                    offset = 0;
+                    msgpack_unpacked_init(&unpacked);
+                    for (index = 0; index < records; index++) {
+                        TEST_CHECK(msgpack_unpack_next(&unpacked, buffer, size, &offset) ==
+                                   MSGPACK_UNPACK_SUCCESS);
+                    }
+                    TEST_CHECK(offset == size);
+                    msgpack_unpacked_destroy(&unpacked);
+                }
+                flb_free(buffer);
+                flb_sds_destroy(json);
+            }
+        }
+    }
+}
+#endif
+
 TEST_LIST = {
+#ifdef FLB_HAVE_YYJSON
+    { "json_pack_depth_limit", test_json_pack_depth_limit },
+#endif
     /* JSON maps iteration */
     { "json_pack"          , test_json_pack },
     { "json_pack_ext_default_backend", test_json_pack_ext_default_backend },