diff --git a/src/flb_pack.c b/src/flb_pack.c index 5127206a2e0..b49217b4006 100644 --- a/src/flb_pack.c +++ b/src/flb_pack.c @@ -43,6 +43,7 @@ #include #include +#include #include #include #ifdef FLB_HAVE_YYJSON @@ -51,6 +52,12 @@ #define try_to_write_str flb_utils_write_str +/* + * Keep traversal state on the heap: output coroutines have small stacks. + * Preserve the nesting limit and truncate deeper containers to JSON null. + */ +#define FLB_PACK_JSON_MAX_DEPTH 512 + static int convert_nan_to_null = FLB_FALSE; static int flb_pack_set_null_as_nan(int b) { @@ -314,61 +321,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); + return msgpack_pack_true(pck); } - else { - msgpack_pack_false(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 +410,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 +489,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); @@ -941,7 +957,7 @@ static inline int try_to_write(char *buf, int *off, size_t left, if (str_len <= 0){ str_len = strlen(str); } - if (left <= *off+str_len) { + if (left < *off + str_len) { return FLB_FALSE; } memcpy(buf+*off, str, str_len); @@ -973,7 +989,9 @@ static inline int key_exists_in_map(msgpack_object key, msgpack_object map, int continue; } - if (memcmp(key.via.str.ptr, p.via.str.ptr, p.via.str.size) == 0) { + if (p.via.str.size == 0 || + (p.via.str.ptr != NULL && + memcmp(key.via.str.ptr, p.via.str.ptr, p.via.str.size) == 0)) { return FLB_TRUE; } } @@ -981,14 +999,27 @@ static inline int key_exists_in_map(msgpack_object key, msgpack_object map, int return FLB_FALSE; } -static int msgpack2json(char *buf, int *off, size_t left, - const msgpack_object *o, int escape_unicode) +/* + * Log the maximum-nesting-depth truncation warning at most once per + * top-level msgpack2json() conversion, regardless of how many separate + * branches in the structure end up being truncated. + */ +static void msgpack2json_depth_warn(int *warned) +{ + if (warned != NULL && *warned == FLB_FALSE) { + flb_warn("[pack] msgpack to JSON conversion exceeded the maximum " + "nesting depth (%d), truncating remaining structure", + FLB_PACK_JSON_MAX_DEPTH); + *warned = FLB_TRUE; + } +} + +static int msgpack2json_scalar(char *buf, int *off, size_t left, + const msgpack_object *o, int escape_unicode) { int i; - int dup; int ret = FLB_FALSE; int loop; - int packed; switch(o->type) { case MSGPACK_OBJECT_NIL: @@ -1034,6 +1065,10 @@ static int msgpack2json(char *buf, int *off, size_t left, break; case MSGPACK_OBJECT_STR: + if (o->via.str.size != 0 && o->via.str.ptr == NULL) { + errno = EINVAL; + return -1; + } if (try_to_write(buf, off, left, "\"", 1) && (o->via.str.size > 0 ? try_to_write_str(buf, off, left, o->via.str.ptr, o->via.str.size, escape_unicode) @@ -1044,6 +1079,10 @@ static int msgpack2json(char *buf, int *off, size_t left, break; case MSGPACK_OBJECT_BIN: + if (o->via.bin.size != 0 && o->via.bin.ptr == NULL) { + errno = EINVAL; + return -1; + } if (try_to_write(buf, off, left, "\"", 1) && (o->via.bin.size > 0 ? try_to_write_str(buf, off, left, o->via.bin.ptr, o->via.bin.size, escape_unicode) @@ -1054,6 +1093,11 @@ static int msgpack2json(char *buf, int *off, size_t left, break; case MSGPACK_OBJECT_EXT: + if (o->via.ext.size > INT_MAX || + (o->via.ext.size != 0 && o->via.ext.ptr == NULL)) { + errno = EINVAL; + return -1; + } if (!try_to_write(buf, off, left, "\"", 1)) { goto msg2json_end; } @@ -1075,73 +1119,181 @@ static int msgpack2json(char *buf, int *off, size_t left, ret = FLB_TRUE; break; - case MSGPACK_OBJECT_ARRAY: - loop = o->via.array.size; + default: + flb_warn("[%s] unknown msgpack type %i", __FUNCTION__, o->type); + errno = EINVAL; + return -1; + } - if (!try_to_write(buf, off, left, "[", 1)) { - goto msg2json_end; - } - if (loop != 0) { - msgpack_object* p = o->via.array.ptr; - if (!msgpack2json(buf, off, left, p, escape_unicode)) { - goto msg2json_end; + msg2json_end: + return ret; +} + +struct msgpack_json_frame { + const msgpack_object *object; + uint32_t index; + int packed; + int value; +}; + +/* 1: complete, 0: output buffer too small, -1: terminal conversion error. */ +static int msgpack2json(char *buf, int *off, size_t left, + const msgpack_object *o, int escape_unicode) +{ + struct msgpack_json_frame *frames; + struct msgpack_json_frame *frame; + const msgpack_object *container; + const msgpack_object *key; + uint32_t count; + int depth = 0; + int warned = FLB_FALSE; + int quote_key; + int ret; + + if (o->type != MSGPACK_OBJECT_ARRAY && o->type != MSGPACK_OBJECT_MAP) { + return msgpack2json_scalar(buf, off, left, o, escape_unicode); + } + + frames = flb_calloc(FLB_PACK_JSON_MAX_DEPTH, sizeof(*frames)); + if (frames == NULL) { + errno = ENOMEM; + return -1; + } + + while (1) { + if (o->type == MSGPACK_OBJECT_ARRAY || o->type == MSGPACK_OBJECT_MAP) { + count = o->type == MSGPACK_OBJECT_ARRAY ? o->via.array.size : o->via.map.size; + if (count > INT_MAX || + (count != 0 && o->type == MSGPACK_OBJECT_ARRAY && o->via.array.ptr == NULL) || + (count != 0 && o->type == MSGPACK_OBJECT_MAP && o->via.map.ptr == NULL)) { + errno = EINVAL; + ret = -1; + break; } - for (i=1; itype == MSGPACK_OBJECT_ARRAY ? "[" : "{", 1); + if (ret == 0) { + break; + } + if (count != 0) { + frame = &frames[depth++]; + frame->object = o; + frame->index = 0; + frame->packed = 0; + frame->value = FLB_FALSE; + goto next_child; } + ret = try_to_write(buf, off, left, + o->type == MSGPACK_OBJECT_ARRAY ? "]" : "}", 1); } } - - ret = try_to_write(buf, off, left, "]", 1); - break; - - case MSGPACK_OBJECT_MAP: - loop = o->via.map.size; - if (!try_to_write(buf, off, left, "{", 1)) { - goto msg2json_end; + else { + quote_key = depth > 0 && frames[depth - 1].object->type == MSGPACK_OBJECT_MAP && + !frames[depth - 1].value && o->type != MSGPACK_OBJECT_STR && + o->type != MSGPACK_OBJECT_BIN && o->type != MSGPACK_OBJECT_EXT; + if (quote_key && !try_to_write(buf, off, left, "\"", 1)) { + ret = 0; + break; + } + ret = msgpack2json_scalar(buf, off, left, o, escape_unicode); + if (ret > 0 && quote_key) { + ret = try_to_write(buf, off, left, "\"", 1); + } + } + if (ret <= 0) { + break; } - if (loop != 0) { - msgpack_object k; - msgpack_object_kv *p = o->via.map.ptr; - - packed = 0; - dup = FLB_FALSE; - k = o->via.map.ptr[0].key; - for (i = 0; i < loop; i++) { - k = o->via.map.ptr[i].key; - dup = key_exists_in_map(k, *o, i + 1); - if (dup == FLB_TRUE) { - continue; + /* Resume the parent without consuming another C stack frame. */ + while (depth > 0) { + frame = &frames[depth - 1]; + if (frame->object->type == MSGPACK_OBJECT_MAP && !frame->value) { + ret = try_to_write(buf, off, left, ":", 1); + if (ret == 0) { + goto done; } - - if (packed > 0) { - if (!try_to_write(buf, off, left, ",", 1)) { - goto msg2json_end; + frame->value = FLB_TRUE; + o = &frame->object->via.map.ptr[frame->index].val; + goto visit; + } + frame->index++; + frame->packed++; + frame->value = FLB_FALSE; + +next_child: + container = frame->object; + count = container->type == MSGPACK_OBJECT_ARRAY ? + container->via.array.size : container->via.map.size; + if (container->type == MSGPACK_OBJECT_MAP) { + while (frame->index < count) { + key = &container->via.map.ptr[frame->index].key; + if (key->type == MSGPACK_OBJECT_ARRAY || key->type == MSGPACK_OBJECT_MAP) { + errno = EINVAL; + ret = -1; + goto done; + } + if (key->type == MSGPACK_OBJECT_STR && key->via.str.size != 0 && + key->via.str.ptr == NULL) { + errno = EINVAL; + ret = -1; + goto done; } + if (!key_exists_in_map(*key, *container, frame->index + 1)) { + break; + } + frame->index++; } - - if ( - !msgpack2json(buf, off, left, &(p+i)->key, escape_unicode) || - !try_to_write(buf, off, left, ":", 1) || - !msgpack2json(buf, off, left, &(p+i)->val, escape_unicode) ) { - goto msg2json_end; + } + if (frame->index < count) { + if (frame->packed > 0 && !try_to_write(buf, off, left, ",", 1)) { + ret = 0; + goto done; } - packed++; + o = container->type == MSGPACK_OBJECT_ARRAY ? + &container->via.array.ptr[frame->index] : + &container->via.map.ptr[frame->index].key; + goto visit; } + ret = try_to_write(buf, off, left, + container->type == MSGPACK_OBJECT_ARRAY ? "]" : "}", 1); + if (ret == 0) { + goto done; + } + depth--; } - - ret = try_to_write(buf, off, left, "}", 1); break; +visit: + continue; + } +done: + flb_free(frames); + return ret; +} - default: - flb_warn("[%s] unknown msgpack type %i", __FUNCTION__, o->type); +/* + * Internal variant used by the growing-buffer wrappers: characters written + * on success, zero if the buffer is too small, or a negative value for a + * terminal conversion error. + */ +static int msgpack_to_json_buf(char *json_str, size_t json_size, + const msgpack_object *obj, int escape_unicode) +{ + int ret = -1; + int off = 0; + + if (json_str == NULL || obj == NULL || json_size == 0 || json_size > INT_MAX) { + errno = EINVAL; + return -1; } - msg2json_end: - return ret; + ret = msgpack2json(json_str, &off, json_size - 1, obj, escape_unicode); + json_str[off] = '\0'; + return ret > 0 ? off : ret; } /** @@ -1152,32 +1304,15 @@ static int msgpack2json(char *buf, int *off, size_t left, * @param json_size The size of json_str. * @param data The msgpack_unpacked data. * @return success ? a number characters filled : negative value + * (including when the buffer is too small) */ int flb_msgpack_to_json(char *json_str, size_t json_size, const msgpack_object *obj, int escape_unicode) { - int ret = -1; - int off = 0; - - if (json_str == NULL || obj == NULL) { - return -1; - } - - /* a zero-capacity buffer would wrap 'json_size - 1' to SIZE_MAX below */ - if (json_size == 0) { - return -1; - } - - ret = msgpack2json(json_str, &off, json_size - 1, obj, escape_unicode); - json_str[off] = '\0'; + int ret; - /* - * msgpack2json() returns FLB_FALSE (0) when the output did not fit in the - * buffer. Report that as a negative value so fixed-buffer callers can - * distinguish truncation from a successful zero-length write, matching the - * documented contract above. - */ - return ret ? off : -1; + ret = msgpack_to_json_buf(json_str, json_size, obj, escape_unicode); + return ret > 0 ? ret : -1; } flb_sds_t flb_msgpack_raw_to_json_sds(const void *in_buf, size_t in_size, int escape_unicode) @@ -1215,8 +1350,13 @@ flb_sds_t flb_msgpack_raw_to_json_sds(const void *in_buf, size_t in_size, int es root = &result.data; while (1) { - ret = flb_msgpack_to_json(out_buf, out_size, root, escape_unicode); - if (ret <= 0) { + ret = msgpack_to_json_buf(out_buf, out_size, root, escape_unicode); + if (ret < 0) { + flb_sds_destroy(out_buf); + msgpack_unpacked_destroy(&result); + return NULL; + } + if (ret == 0) { realloc_size *= 2; tmp_buf = flb_sds_increase(out_buf, realloc_size); if (tmp_buf) { @@ -1640,8 +1780,12 @@ char *flb_msgpack_to_json_str(size_t size, const msgpack_object *obj, int escape } while (1) { - ret = flb_msgpack_to_json(buf, size, obj, escape_unicode); - if (ret <= 0) { + ret = msgpack_to_json_buf(buf, size, obj, escape_unicode); + if (ret < 0) { + flb_free(buf); + return NULL; + } + if (ret == 0) { /* buffer is small. retry.*/ size *= 2; tmp = flb_realloc(buf, size); diff --git a/tests/integration/scenarios/out_loki/config/out_loki_long_unicode.yaml b/tests/integration/scenarios/out_loki/config/out_loki_long_unicode.yaml index 4bd4d59c78c..d3d0ce7eb74 100644 --- a/tests/integration/scenarios/out_loki/config/out_loki_long_unicode.yaml +++ b/tests/integration/scenarios/out_loki/config/out_loki_long_unicode.yaml @@ -1,4 +1,5 @@ service: + coro_stack_size: 24576 flush: 1 grace: 1 log_level: info diff --git a/tests/integration/scenarios/out_loki/tests/test_out_loki_001.py b/tests/integration/scenarios/out_loki/tests/test_out_loki_001.py index 63a52f0082b..8155d0c2309 100644 --- a/tests/integration/scenarios/out_loki/tests/test_out_loki_001.py +++ b/tests/integration/scenarios/out_loki/tests/test_out_loki_001.py @@ -131,3 +131,29 @@ def test_out_loki_preserves_long_unicode_json_strings(): assert len(records) == RECORD_COUNT assert all(set(record) == {"msg"} for record in records) assert sorted(record["msg"] for record in records) == sorted(expected_messages) + + +def test_out_loki_preserves_nested_records_on_small_stack(): + nested = {"leaf": "preserved"} + # Each iteration adds two containers. Leave room for the leaf, record map, + # and log-event envelope within the system MessagePack limit of 32 levels. + for _ in range(14): + nested = {"child": [nested]} + + with tempfile.TemporaryDirectory(prefix="flb-out-loki-depth-") as temp_directory: + input_path = Path(temp_directory) / "input.log" + with input_path.open("w", encoding="utf-8") as input_file: + for sequence in range(RECORD_COUNT): + json.dump({"seq": sequence, "msg": nested}, input_file) + input_file.write("\n") + service = Service(input_path) + try: + service.start() + service.wait_for_records() + finally: + service.stop() + records = collect_loki_records() + + assert len(records) == RECORD_COUNT + for record in records: + assert record == {"msg": nested} 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..6557555cc6e --- /dev/null +++ b/tests/integration/scenarios/security_ingestors/tests/test_security_ingestors.py @@ -0,0 +1,147 @@ +"""Bounded nesting and recovery checks for network ingestion.""" +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" + + +@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") + input_config = {"name": plugin} + if plugin == "unix_socket": + input_config["socket_path"] = address + else: + input_config.update({"listen": "127.0.0.1", "port": port}) + if mode.endswith("-parser"): + input_config.update({"format": "none", "parser": "json"}) + if plugin == "syslog": + input_config.update({"mode": "tcp", "parser": "json"}) + config = { + "service": { + "flush": 0.1, + "grace": 1, + "parsers_file": str(parser), + "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 address, manager.process, log + finally: + manager.stop() + + +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 2758afb7462..bb475b0aff6 100644 --- a/tests/internal/pack.c +++ b/tests/internal/pack.c @@ -6,7 +6,9 @@ #include #include #include +#include #include +#include #include #include @@ -1118,6 +1120,160 @@ void test_json_pack_bug5336() } /* Ensure empty arrays inside nested objects are handled */ +/* + * Must mirror FLB_PACK_JSON_MAX_DEPTH in src/flb_pack.c (not exposed via a + * public header since it is an internal implementation detail of + * msgpack2json()). + */ +#define TEST_PACK_MAX_DEPTH 512 + +/* + * Regression test: a non-empty msgpack map whose key/value pairs would be + * evaluated exactly one level past FLB_PACK_JSON_MAX_DEPTH must still be + * rendered as valid JSON. The whole map is expected to collapse to a JSON + * null literal instead of emitting a bare, unquoted null in place of a map + * key (which would produce invalid JSON such as {null:"v"}). + */ +void test_json_pack_deep_map_boundary() +{ + int i; + char *out; + char *p; + msgpack_object *chain; + msgpack_object_kv kv; + size_t out_len; + size_t expected_len; + + /* + * Build TEST_PACK_MAX_DEPTH nested single-element arrays with a + * non-empty map ({"k":"v"}) as the innermost element, constructed + * directly in memory. chain[0] is evaluated at depth 0, chain[i] at + * depth i, so the map at chain[TEST_PACK_MAX_DEPTH] is evaluated at + * depth == TEST_PACK_MAX_DEPTH: its own guard passes, but its key/value + * pair would be one level past the limit. + */ + chain = flb_malloc(sizeof(msgpack_object) * (TEST_PACK_MAX_DEPTH + 1)); + if (!TEST_CHECK(chain != NULL)) { + TEST_MSG("could not allocate test msgpack_object chain"); + return; + } + + for (i = 0; i < TEST_PACK_MAX_DEPTH; i++) { + chain[i].type = MSGPACK_OBJECT_ARRAY; + chain[i].via.array.size = 1; + chain[i].via.array.ptr = &chain[i + 1]; + } + + kv.key.type = MSGPACK_OBJECT_STR; + kv.key.via.str.size = 1; + kv.key.via.str.ptr = "k"; + kv.val.type = MSGPACK_OBJECT_STR; + kv.val.via.str.size = 1; + kv.val.via.str.ptr = "v"; + + chain[TEST_PACK_MAX_DEPTH].type = MSGPACK_OBJECT_MAP; + chain[TEST_PACK_MAX_DEPTH].via.map.size = 1; + chain[TEST_PACK_MAX_DEPTH].via.map.ptr = &kv; + + out = flb_msgpack_to_json_str(1024, &chain[0], FLB_FALSE); + flb_free(chain); + + if (!TEST_CHECK(out != NULL)) { + TEST_MSG("flb_msgpack_to_json_str returned NULL"); + return; + } + + /* a map key must never be truncated to an unquoted null */ + p = strstr(out, "null:"); + if (!TEST_CHECK(p == NULL)) { + TEST_MSG("map key was rendered as an unquoted null: %s", out); + } + + /* + * Exact shape check: TEST_PACK_MAX_DEPTH opening brackets, then the + * truncated map as a bare "null", then TEST_PACK_MAX_DEPTH closing + * brackets. Verify both the total length and that the null literal + * begins exactly at offset TEST_PACK_MAX_DEPTH, rather than accepting + * "null" anywhere in the output. + */ + { + out_len = strlen(out); + expected_len = (size_t) TEST_PACK_MAX_DEPTH * 2 + 4; + + if (!TEST_CHECK(out_len == expected_len)) { + TEST_MSG("unexpected output length: expected=%zu got=%zu out=%s", + expected_len, out_len, out); + } + + if (!TEST_CHECK(out_len > (size_t) TEST_PACK_MAX_DEPTH + 4 && + strncmp(out + TEST_PACK_MAX_DEPTH, "null", 4) == 0)) { + TEST_MSG("expected a null literal at offset %d: %s", + TEST_PACK_MAX_DEPTH, out); + } + } + + /* the original key/value content must not appear: it was truncated */ + p = strstr(out, "\"k\":\"v\""); + if (!TEST_CHECK(p == NULL)) { + TEST_MSG("map content should have been truncated: %s", out); + } + + flb_free(out); +} + +/* + * Companion check: the same map shape placed comfortably below the depth + * limit must still serialize its real content (i.e. the pre-check added for + * the boundary case above must not fire early for valid, shallower input). + */ +void test_json_pack_deep_map_below_boundary() +{ + int i; + int shallow_depth = TEST_PACK_MAX_DEPTH - 5; + char *out; + char *p; + msgpack_object *chain; + msgpack_object_kv kv; + + chain = flb_malloc(sizeof(msgpack_object) * (shallow_depth + 1)); + if (!TEST_CHECK(chain != NULL)) { + TEST_MSG("could not allocate test msgpack_object chain"); + return; + } + + for (i = 0; i < shallow_depth; i++) { + chain[i].type = MSGPACK_OBJECT_ARRAY; + chain[i].via.array.size = 1; + chain[i].via.array.ptr = &chain[i + 1]; + } + + kv.key.type = MSGPACK_OBJECT_STR; + kv.key.via.str.size = 1; + kv.key.via.str.ptr = "k"; + kv.val.type = MSGPACK_OBJECT_STR; + kv.val.via.str.size = 1; + kv.val.via.str.ptr = "v"; + + chain[shallow_depth].type = MSGPACK_OBJECT_MAP; + chain[shallow_depth].via.map.size = 1; + chain[shallow_depth].via.map.ptr = &kv; + + out = flb_msgpack_to_json_str(1024, &chain[0], FLB_FALSE); + flb_free(chain); + + if (!TEST_CHECK(out != NULL)) { + TEST_MSG("flb_msgpack_to_json_str returned NULL"); + return; + } + + p = strstr(out, "\"k\":\"v\""); + if (!TEST_CHECK(p != NULL)) { + TEST_MSG("map content below the depth limit should be preserved: %s", out); + } + + flb_free(out); +} + void test_json_pack_empty_array() { int ret; @@ -1328,7 +1484,401 @@ void test_json_pack_token_count_overflow() flb_pack_state_reset(&state); } + +void test_json_pack_buffer_capacity(void) +{ + struct { + const char *packed; + size_t packed_size; + const char *raw; + const char *escaped; + } cases[] = { + {"\xc0", 1, "null", "null"}, + {"\xc3", 1, "true", "true"}, + {"\xc2", 1, "false", "false"}, + {"\x2a", 1, "42", "42"}, + {"\xd0\xd6", 2, "-42", "-42"}, + {"\xcb\x3f\xf8\x00\x00\x00\x00\x00\x00", 9, "1.5", "1.5"}, + {"\xa0", 1, "\"\"", "\"\""}, + {"\xa4" "abcd", 5, "\"abcd\"", "\"abcd\""}, + {"\xa6" "a\n\"\\\xc3\xa9", 7, + "\"a\\n\\\"\\\\\xc3\xa9\"", "\"a\\n\\\"\\\\\\u00e9\""}, + {"\xa4\xf0\x9f\x98\x80", 5, + "\"\xf0\x9f\x98\x80\"", "\"\\ud83d\\ude00\""}, + {"\xd9\x20" "0123456789abcdef0123456789abcdef", 34, + "\"0123456789abcdef0123456789abcdef\"", + "\"0123456789abcdef0123456789abcdef\""}, + {"\xc4\x04" "abcd", 6, "\"abcd\"", "\"abcd\""}, + {"\xd4\x01" "a", 3, "\"\\x61\"", "\"\\x61\""}, + {"\x90", 1, "[]", "[]"}, + {"\x80", 1, "{}", "{}"}, + {"\x81\xa1" "a" "\x93\xc0\x81\xa1" "b" "\xa4" "abcd" "\x90", 14, + "{\"a\":[null,{\"b\":\"abcd\"},[]]}", + "{\"a\":[null,{\"b\":\"abcd\"},[]]}"}, + }; + size_t i; + size_t capacity; + size_t expected_size; + size_t offset; + int escape_unicode; + int ret; + char *allocation; + char *buffer; + const char *expected; + msgpack_unpacked unpacked; + + for (i = 0; i < sizeof(cases) / sizeof(cases[0]); i++) { + offset = 0; + msgpack_unpacked_init(&unpacked); + ret = msgpack_unpack_next(&unpacked, cases[i].packed, + cases[i].packed_size, &offset); + TEST_ASSERT(ret == MSGPACK_UNPACK_SUCCESS); + TEST_CHECK(offset == cases[i].packed_size); + + for (escape_unicode = FLB_FALSE; escape_unicode <= FLB_TRUE; escape_unicode++) { + expected = escape_unicode ? cases[i].escaped : cases[i].raw; + expected_size = strlen(expected); + + for (capacity = 0; capacity <= expected_size + 2; capacity++) { + allocation = flb_malloc(capacity + 2); + TEST_ASSERT(allocation != NULL); + memset(allocation, 'X', capacity + 2); + buffer = allocation + 1; + + ret = flb_msgpack_to_json(buffer, capacity, &unpacked.data, + escape_unicode); + TEST_CHECK(allocation[0] == 'X'); + TEST_CHECK(buffer[capacity] == 'X'); + TEST_MSG("case=%zu capacity=%zu escape_unicode=%d", + i, capacity, escape_unicode); + if (capacity == 0) { + TEST_CHECK(ret == -1); + } + else if (capacity <= expected_size) { + TEST_CHECK(ret < 0); + TEST_CHECK(memchr(buffer, '\0', capacity) != NULL); + } + else { + TEST_CHECK(ret == expected_size); + TEST_MSG("case=%zu capacity=%zu escape_unicode=%d ret=%d", + i, capacity, escape_unicode, ret); + TEST_CHECK(memcmp(buffer, expected, expected_size + 1) == 0); + } + flb_free(allocation); + } + } + msgpack_unpacked_destroy(&unpacked); + } +} + +#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 + +static cothread_t pack_coro_caller; +static msgpack_object *pack_coro_object; +static char *pack_coro_json; + +static void pack_small_stack_entry(void) +{ + pack_coro_json = flb_msgpack_to_json_str(1, pack_coro_object, FLB_FALSE); + co_switch(pack_coro_caller); +} + +/* Match the small coroutine stacks used by older Linux output workers. */ +static void check_json_pack_small_stack(int depth, int mixed) +{ + msgpack_object *chain; + msgpack_object_kv *pairs; + cothread_t callee; + char *expected; + int offset = 0; + int retained; + size_t stack_size; +#ifdef FLB_HAVE_VALGRIND + unsigned int stack_id; +#endif + int i; + + chain = flb_calloc(depth + 1, sizeof(*chain)); + TEST_ASSERT(chain != NULL); + pairs = flb_calloc(depth, sizeof(*pairs)); + TEST_ASSERT(pairs != NULL); + expected = flb_malloc(depth * 6 + 5); + TEST_ASSERT(expected != NULL); + retained = depth > 512 ? 512 : depth; + for (i = 0; i < depth; i++) { + if (mixed && i % 2 == 0) { + chain[i].type = MSGPACK_OBJECT_MAP; + chain[i].via.map.size = 1; + chain[i].via.map.ptr = &pairs[i]; + pairs[i].key.type = MSGPACK_OBJECT_STR; + pairs[i].key.via.str.ptr = "k"; + pairs[i].key.via.str.size = 1; + } + else { + chain[i].type = MSGPACK_OBJECT_ARRAY; + chain[i].via.array.size = 1; + chain[i].via.array.ptr = &chain[i + 1]; + } + if (i < retained) { + offset += sprintf(expected + offset, mixed && i % 2 == 0 ? "{\"k\":" : "["); + } + } + chain[depth].type = MSGPACK_OBJECT_NIL; + for (i = 0; i < depth; i++) { + if (mixed && i % 2 == 0) { + pairs[i].val = chain[i + 1]; + } + } + offset += sprintf(expected + offset, "null"); + for (i = retained - 1; i >= 0; i--) { + expected[offset++] = mixed && i % 2 == 0 ? '}' : ']'; + } + expected[offset] = '\0'; + pack_coro_object = chain; + pack_coro_json = NULL; + pack_coro_caller = co_active(); + callee = co_create(24576, pack_small_stack_entry, &stack_size); + TEST_ASSERT(callee != NULL); +#ifdef FLB_HAVE_VALGRIND + stack_id = VALGRIND_STACK_REGISTER(callee, ((char *) callee) + stack_size); +#endif + co_switch(callee); +#ifdef FLB_HAVE_VALGRIND + VALGRIND_STACK_DEREGISTER(stack_id); +#endif + co_delete(callee); + TEST_CHECK(pack_coro_json != NULL); + if (pack_coro_json != NULL) { + TEST_CHECK(strcmp(pack_coro_json, expected) == 0); + flb_free(pack_coro_json); + } + flb_free(chain); + flb_free(pairs); + flb_free(expected); +} + +void test_json_pack_small_stack(void) +{ + check_json_pack_small_stack(200, FLB_FALSE); + check_json_pack_small_stack(200, FLB_TRUE); + check_json_pack_small_stack(2000, FLB_TRUE); +} + +void test_json_pack_scalar_keys(void) +{ + const char packed[] = "\x86\x01\xc0\xd0\xd6\xc3\xc3\x92\x01\xc2\xc2\x80" + "\xc0\xa1n\xcb\x3f\xf8\x00\x00\x00\x00\x00\x00" + "\x81\xa1k\xc2"; + const char *expected = "{\"1\":null,\"-42\":true,\"true\":[1,false],\"false\":{}," + "\"null\":\"n\",\"1.5\":{\"k\":false}}"; + msgpack_unpacked unpacked; + size_t offset = 0; + size_t capacity; + size_t expected_size; + int escape_unicode; + int ret; + char buffer[256]; + char *json; + flb_sds_t sds; + + msgpack_unpacked_init(&unpacked); + ret = msgpack_unpack_next(&unpacked, packed, sizeof(packed) - 1, &offset); + TEST_ASSERT(ret == MSGPACK_UNPACK_SUCCESS); + expected_size = strlen(expected); + + for (escape_unicode = FLB_FALSE; escape_unicode <= FLB_TRUE; escape_unicode++) { + for (capacity = 0; capacity <= expected_size + 1; capacity++) { + memset(buffer, 'X', sizeof(buffer)); + ret = flb_msgpack_to_json(buffer + 1, capacity, &unpacked.data, escape_unicode); + TEST_CHECK(buffer[0] == 'X'); + TEST_CHECK(buffer[capacity + 1] == 'X'); + if (capacity <= expected_size) { + TEST_CHECK(ret < 0); + } + else { + TEST_CHECK(ret == expected_size); + TEST_CHECK(strcmp(buffer + 1, expected) == 0); + } + } + + json = flb_msgpack_to_json_str(1, &unpacked.data, escape_unicode); + TEST_CHECK(json != NULL); + if (json != NULL) { + TEST_CHECK(strcmp(json, expected) == 0); + flb_free(json); + } + sds = flb_msgpack_raw_to_json_sds(packed, sizeof(packed) - 1, escape_unicode); + TEST_CHECK(sds != NULL); + if (sds != NULL) { + TEST_CHECK(strcmp(sds, expected) == 0); + flb_sds_destroy(sds); + } + } + msgpack_unpacked_destroy(&unpacked); +} + +void test_json_pack_terminal_error(void) +{ + msgpack_object object; + msgpack_object_kv pairs[2]; + char *json; + char buffer[16]; + const char invalid_key[] = {0x81, 0x90, 0xc0}; + const char invalid_map_key[] = {0x81, 0x80, 0xc0}; + + memset(&object, 0, sizeof(object)); + object.type = MSGPACK_OBJECT_STR; + object.via.str.size = 1; + object.via.str.ptr = NULL; + TEST_CHECK(flb_msgpack_to_json(buffer, sizeof(buffer), &object, FLB_FALSE) == -1); + json = flb_msgpack_to_json_str(1, &object, FLB_FALSE); + TEST_CHECK(json == NULL); + flb_free(json); + + /* A later invalid key must not be dereferenced during duplicate lookup. */ + memset(pairs, 0, sizeof(pairs)); + pairs[0].key.type = MSGPACK_OBJECT_STR; + pairs[0].key.via.str.ptr = "k"; + pairs[0].key.via.str.size = 1; + pairs[0].val.type = MSGPACK_OBJECT_NIL; + pairs[1] = pairs[0]; + pairs[1].key.via.str.ptr = NULL; + object.type = MSGPACK_OBJECT_MAP; + object.via.map.ptr = pairs; + object.via.map.size = 2; + json = flb_msgpack_to_json_str(1, &object, FLB_FALSE); + TEST_CHECK(json == NULL); + flb_free(json); + + /* Both allocating wrappers must stop on terminal conversion errors. */ + json = flb_msgpack_raw_to_json_sds(invalid_key, sizeof(invalid_key), FLB_FALSE); + TEST_CHECK(json == NULL); + flb_sds_destroy(json); + + json = flb_msgpack_raw_to_json_sds(invalid_map_key, sizeof(invalid_map_key), FLB_FALSE); + TEST_CHECK(json == NULL); + flb_sds_destroy(json); +} + +void test_json_pack_iterative_buffer_retry(void) +{ + const char *input = "{\"dup\":0,\"a\":[{},[],{\"b\":[true,null,\"x\\ny\"]}]," + "\"dup\":2,\"last\":{\"empty\":[]}}"; + const char *expected = "{\"a\":[{},[],{\"b\":[true,null,\"x\\ny\"]}]," + "\"dup\":2,\"last\":{\"empty\":[]}}"; + char *packed; + char *json; + char buffer[256]; + size_t packed_size; + size_t offset = 0; + size_t capacity; + int root_type; + int ret; + msgpack_unpacked result; + + ret = flb_pack_json(input, strlen(input), &packed, &packed_size, &root_type, NULL); + TEST_ASSERT(ret == 0); + msgpack_unpacked_init(&result); + ret = msgpack_unpack_next(&result, packed, packed_size, &offset); + TEST_ASSERT(ret == MSGPACK_UNPACK_SUCCESS); + + /* Exercise early exits at every bracket, key, separator and scalar. */ + for (capacity = 0; capacity < strlen(expected) + 3; capacity++) { + memset(buffer, 'X', sizeof(buffer)); + ret = flb_msgpack_to_json(buffer, capacity, &result.data, FLB_FALSE); + TEST_CHECK(buffer[capacity] == 'X'); + if (capacity == 0) { + TEST_CHECK(ret == -1); + } + else if (ret > 0) { + TEST_CHECK(strcmp(buffer, expected) == 0); + } + else { + /* insufficient space must be reported as a negative value */ + TEST_CHECK(ret < 0); + } + } + json = flb_msgpack_to_json_str(1, &result.data, FLB_FALSE); + TEST_ASSERT(json != NULL); + TEST_CHECK(strcmp(json, expected) == 0); + flb_free(json); + msgpack_unpacked_destroy(&result); + flb_free(packed); +} + TEST_LIST = { + { "json_pack_buffer_capacity", test_json_pack_buffer_capacity }, +#ifdef FLB_HAVE_YYJSON + { "json_pack_depth_limit", test_json_pack_depth_limit }, +#endif + + { "json_pack_small_stack", test_json_pack_small_stack }, + { "json_pack_scalar_keys", test_json_pack_scalar_keys }, + { "json_pack_terminal_error", test_json_pack_terminal_error }, + { "json_pack_iterative_buffer_retry", test_json_pack_iterative_buffer_retry }, /* JSON maps iteration */ { "json_pack" , test_json_pack }, { "json_pack_ext_default_backend", test_json_pack_ext_default_backend }, @@ -1347,6 +1897,8 @@ TEST_LIST = { { "json_pack_truncation", test_json_pack_truncation}, { "json_pack_bug5336" , test_json_pack_bug5336}, { "json_pack_empty_array", test_json_pack_empty_array}, + { "json_pack_deep_map_boundary", test_json_pack_deep_map_boundary}, + { "json_pack_deep_map_below_boundary", test_json_pack_deep_map_below_boundary}, { "json_date_iso8601" , test_json_date_iso8601}, { "json_date_double" , test_json_date_double}, { "json_date_java_sql" , test_json_date_java_sql},