diff --git a/src/flb_pack.c b/src/flb_pack.c index 581255f328b..3542aaae386 100644 --- a/src/flb_pack.c +++ b/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); diff --git a/tests/integration/scenarios/security_ingestors/tests/test_security_ingestors.py b/tests/integration/scenarios/security_ingestors/tests/test_security_ingestors.py new file mode 100644 index 00000000000..0cbfcfc4389 --- /dev/null +++ b/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() diff --git a/tests/internal/pack.c b/tests/internal/pack.c index 60f79238286..371649d75c6 100644 --- a/tests/internal/pack.c +++ b/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 },