diff --git a/plugins/processor_fts/README.md b/plugins/processor_fts/README.md index ff3bf081f2f..c4d20dcd02b 100644 --- a/plugins/processor_fts/README.md +++ b/plugins/processor_fts/README.md @@ -184,6 +184,12 @@ order is preserved. | `fb.set_severity(log, level)` | Set severity text and number from a level name: `trace`, `debug`, `info`, `warn`, `error`, `fatal`, and common aliases. | | `fb.set_trace_context(log, trace_hex, span_hex)` | Set the trace and span IDs from hexadecimal strings. IDs must be nonzero. | | `fb.log_field(log, name)` | Read an OpenTelemetry log field such as `severity_number`, `trace_id`, or `observed_timestamp`. | +| `fb.unpack_json(log, field)` | Decode a JSON object from a body field (`message` by default) and replace `log.body`. Return `true` on success or `false, error` for a missing/non-string field, ordinary parse failure, or non-object JSON; the original body is preserved for these failures. Uses the same typed parser and limits as `fb.parse_json`. | + +Instruction-limit exhaustion in the JSON helpers fails the callback instead of +returning a recoverable decode error. Changes are rolled +back and `error_mode` determines the record outcome: the default `drop` removes +the record, while `keep` forwards the original. ### Example: parse an embedded JSON message @@ -191,33 +197,69 @@ order is preserved. local fb = require("fluentbit") function process(log) - local text = log.body:get_string("message") - if not text or not text:match("^%s*{") then - return - end + fb.unpack_json(log, "message") + fb.redact(log.body, {"authorization", "password"}) +end +``` - local ok, app = pcall(fb.parse_json, text) - if not ok then - return fb.DROP - end +The example allows mixed streams to pass plain text through. The unpack call +keeps plain text, missing/non-string fields, malformed JSON, and non-object JSON +unchanged. `fb.redact` matches keys in maps and arrays; it does not inspect +secrets inside the unchanged message string. If every forwarded message must be +decoded for key-based redaction, drop records when decoding fails: - fb.redact(app, {"authorization", "password"}) - log.body = app +```lua +function process(log) + if not fb.unpack_json(log, "message") then return fb.DROP end + fb.redact(log.body, {"authorization", "password"}) +end +``` + +### Decode a field in place + +Use `fb.decode_json(container, key)` to replace only the selected string field +with its decoded value. The same helper works with body maps, nested maps, +OpenTelemetry attributes, and array elements: +```lua +function process(log) + fb.decode_json(log.body, "message") if fb.is_otel(log) then - local level = app:get_string("level") - if level then - fb.set_severity(log, level) - end - - local time = app:get_string("time") - if time then - log.timestamp = fb.parse_time(time) - end + fb.decode_json(log.attributes, "payload") + fb.decode_json(log.resource.attributes, "labels") end + fb.redact(log.body, {"authorization", "password"}) end ``` +For example, `{"message":"{\"level\":\"info\"}","host":"web"}` becomes +`{"message":{"level":"info"},"host":"web"}`. By comparison, +`fb.unpack_json(log, "message")` produces `{"level":"info"}` as the whole body. +Both operations decode one selected string; strings inside the decoded JSON +require another explicit call. + +`fb.decode_json` accepts any valid JSON value, including arrays, null, and +booleans. `fb.unpack_json` requires an object because a log body must be a map. +Missing/non-string fields and ordinary parse failures return `false, error` +without changing the target. Successful decoding returns `true`. Invalid +container/key arguments, expired views, and failures while applying a mutation +raise errors and follow the callback's `error_mode`. + +For OpenTelemetry log bodies and attributes, the JSON helpers reject unsigned +integers above `9223372036854775807` (`INT64_MAX`), including values nested in +objects or arrays. OTLP AnyValue integers must fit signed 64 bits. The helpers +return `false, error` before mutation, preserving the original field or body. +Ordinary logs, detached scratch maps, `fb.parse_json`, and `fb.try_parse_json` +retain full unsigned 64-bit support. + +These helper statuses are separate from callback results. Leaving `process` +without an explicit return is equivalent to `return fb.KEEP`: successful +changes are committed and the record is forwarded. To remove a record, return +`fb.DROP` explicitly; do not return a helper's boolean status from `process`. + +Use `fb.try_parse_json(text)` when you need the parsed value without mutating a +container. `fb.parse_json(text)` remains the throwing parser. + ## Metrics `process(metric)` runs once per metric family. Resource and scope changes apply @@ -316,6 +358,8 @@ parent references. | `fb.redact(value, keys, replacement)` | Replace the values of matching keys in nested maps and arrays. Key matching is case-insensitive. The default replacement is `[REDACTED]`. | | `fb.keep_keys(map, keys)` | Remove every key not listed. | | `fb.parse_json(text)` | Parse a JSON document into typed values. Integer widths are preserved; duplicate keys and nesting beyond 64 levels are rejected. Input is limited to 1 MiB. | +| `fb.decode_json(container, key)` | Decode a string field in a typed map or array in place. Return `true` on success or `false, error` for a missing/non-string field or ordinary parse failure. Preserve sibling fields and leave the source unchanged on failure. Array indexes start at 1; map keys are literal. | +| `fb.try_parse_json(text)` | Return the same typed value as `fb.parse_json`, or `nil, error` for invalid JSON or non-string input (including a missing field). JSON `null` returns a typed null value, and JSON `false` returns a typed boolean value. The same parser and policy limits apply. | | `fb.parse_time(text)` | Parse an RFC 3339 timestamp into `{sec, nsec}`. | Key-based redaction does not inspect free-form strings; secrets embedded in diff --git a/plugins/processor_fts/fts.h b/plugins/processor_fts/fts.h index 352ee05cc67..2a62e8a36ee 100644 --- a/plugins/processor_fts/fts.h +++ b/plugins/processor_fts/fts.h @@ -156,6 +156,8 @@ struct fts_runtime { int empty_metatable_ref; int key_refs[FTS_KEY_COUNT]; int tag_ref; + /* Current OTLP event for destination-aware JSON validation. */ + int json_event_ref; uint64_t generation; struct fts_undo_entry *undo; size_t undo_count; diff --git a/plugins/processor_fts/fts_log_helpers.lua b/plugins/processor_fts/fts_log_helpers.lua index f19f9d7a39c..b8e71331d07 100644 --- a/plugins/processor_fts/fts_log_helpers.lua +++ b/plugins/processor_fts/fts_log_helpers.lua @@ -1,4 +1,18 @@ local fb = package.loaded.fluentbit +--- Replace a log body with a decoded object; preserve it and return false, error on failure. +function fb.unpack_json(log, field) + local source = log.body:get(field or 'message') + if not source or source:type() ~= 'string' then + return false, 'JSON source field must be a string' + end + local value, err = fb.try_parse_json(source:value()) + if not value then return false, err end + if value:type() ~= 'map' then return false, 'JSON body must be an object' end + err = fb._fts_json_integer_error(log.body, value) + if err then return false, err end + log.body = value + return true +end function fb.is_otel(log) local schema = rawget(log, 'context') and rawget(log, 'context').metadata:get('schema') return schema and schema:type() == 'string' and schema:value() == 'otlp' or false diff --git a/plugins/processor_fts/fts_logs.c b/plugins/processor_fts/fts_logs.c index 3bd091a5a47..8b9685d6d82 100644 --- a/plugins/processor_fts/fts_logs.c +++ b/plugins/processor_fts/fts_logs.c @@ -289,6 +289,120 @@ static struct cfl_variant *log_field(struct cfl_variant *metadata, const char *k return value ? value : map_field(metadata, key); } +/* Compare live ancestors, including views into a replaced log body. */ +static int json_target_within(lua_State *l, struct cfl_variant *payload) +{ + struct fts_value *value; + int base; + int depth; + + if (!payload) { + return FLB_FALSE; + } + + base = lua_gettop(l); + lua_pushvalue(l, 1); + for (depth = 0; depth < FTS_MAX_DEPTH; depth++) { + if (fts_resolve_value(l, -1, 0) == payload) { + lua_settop(l, base); + return FLB_TRUE; + } + value = lua_touserdata(l, -1); + if (value->owned) { + break; + } + fts_lua_getuservalue(l, -1); + lua_rawgeti(l, -1, 1); + lua_replace(l, base + 1); + lua_settop(l, base + 1); + } + lua_settop(l, base); + return FLB_FALSE; +} + +/* Read a typed field from an already validated table without invoking policy code. */ +static struct cfl_variant *json_payload_field(lua_State *l, int table, const char *key) +{ + struct cfl_variant *value; + + lua_pushstring(l, key); + lua_rawget(l, table); + value = fts_resolve_value(l, -1, 0); + lua_pop(l, 1); + return value; +} + +/* Check every nested integer while charging traversal to the policy budget. */ +static int json_integers_fit(lua_State *l, struct fts_runtime *ctx, + struct cfl_variant *value, int depth) +{ + struct cfl_list *head; + struct cfl_kvpair *pair; + size_t i; + + if (depth >= FTS_MAX_DEPTH || (!ctx->jit && --ctx->instructions_left <= 0)) { + return luaL_error(l, "JSON integer validation exceeds nesting or instruction limit"); + } + if (value->type == CFL_VARIANT_UINT) { + return value->data.as_uint64 <= INT64_MAX; + } + if (value->type == CFL_VARIANT_KVLIST) { + cfl_list_foreach(head, &value->data.as_kvlist->list) { + pair = cfl_list_entry(head, struct cfl_kvpair, _head); + if (!json_integers_fit(l, ctx, pair->val, depth + 1)) { + return FLB_FALSE; + } + } + } + else if (value->type == CFL_VARIANT_ARRAY) { + for (i = 0; i < value->data.as_array->entry_count; i++) { + if (!json_integers_fit(l, ctx, value->data.as_array->entries[i], depth + 1)) { + return FLB_FALSE; + } + } + } + return FLB_TRUE; +} + +/* JSON decode failures must be recoverable before OTLP commit validation. */ +static int json_integer_error(lua_State *l) +{ + struct fts_runtime *ctx; + struct cfl_variant *metadata; + struct cfl_variant *value; + int event; + int payload; + + lua_getallocf(l, (void **) &ctx); + lua_rawgeti(l, LUA_REGISTRYINDEX, ctx->json_event_ref); + if (!lua_istable(l, -1)) { + lua_pushnil(l); + return 1; + } + event = lua_gettop(l); + payload = json_target_within(l, json_payload_field(l, event, "body")); + if (!payload) { + metadata = json_payload_field(l, event, "metadata"); + payload = json_target_within(l, log_field(metadata, "attributes")); + } + if (!payload) { + lua_pushliteral(l, "context"); + lua_rawget(l, event); + if (lua_istable(l, -1)) { + value = json_payload_field(l, lua_gettop(l), "attributes"); + payload = json_target_within(l, value); + } + lua_pop(l, 1); + } + if (payload && !json_integers_fit(l, ctx, fts_resolve_value(l, 2, 0), 0)) { + lua_pushliteral(l, "OTel AnyValue integers must fit signed 64 bits"); + } + else { + lua_pushnil(l); + } + return 1; +} + static void push_changed_fields(lua_State *l, struct record_call *call, struct cfl_variant *metadata, struct cfl_variant *context) { @@ -578,6 +692,7 @@ static void require_plain_table(lua_State *l, int index, const char *name) } } +/* Run one log policy with transactional values and validate its committed result. */ static int run_record(lua_State *l) { struct record_call *call; @@ -649,6 +764,8 @@ static int run_record(lua_State *l) } if (has_otel_context(call->metadata)) { + lua_pushvalue(l, event); + lua_rawseti(l, LUA_REGISTRYINDEX, call->ctx->json_event_ref); lua_rawgeti(l, LUA_REGISTRYINDEX, call->ctx->module_ref); lua_getfield(l, -1, "_attach"); lua_pushvalue(l, event); @@ -1976,8 +2093,22 @@ int fts_process_logs(struct flb_processor_instance *ins, void *data, return FLB_PROCESSOR_FAILURE; } +/* Reserve callback context storage and expose native and Lua log helpers. */ static int register_log_helpers(lua_State *l) { + struct fts_runtime *ctx; + + lua_getallocf(l, (void **) &ctx); + lua_pushboolean(l, 0); + ctx->json_event_ref = luaL_ref(l, LUA_REGISTRYINDEX); + + lua_getglobal(l, "package"); + lua_getfield(l, -1, "loaded"); + lua_getfield(l, -1, "fluentbit"); + lua_pushcfunction(l, json_integer_error); + lua_setfield(l, -2, "_fts_json_integer_error"); + lua_pop(l, 3); + if (fts_load_helpers(l, fts_log_helpers, sizeof(fts_log_helpers) - 1, "fluentbit.log_helpers") != 0) { return -1; diff --git a/plugins/processor_fts/fts_runtime.c b/plugins/processor_fts/fts_runtime.c index 32722349090..5fb1cbbbb48 100644 --- a/plugins/processor_fts/fts_runtime.c +++ b/plugins/processor_fts/fts_runtime.c @@ -363,12 +363,14 @@ static int initialize_policy(lua_State *l) return 0; } +/* Initialize the bounded Lua runtime and the selected signal adapter's helpers. */ int fts_runtime_initialize(struct fts_runtime *ctx, struct flb_config *config) { lua_State *l; int ret; ctx->env = config->env; + ctx->json_event_ref = LUA_NOREF; ctx->vm = flb_luajit_create(config); if (!ctx->vm) { return FLB_PROCESSOR_FAILURE; @@ -451,6 +453,11 @@ int fts_runtime_call(struct fts_runtime *ctx, void *data, int results) if (ret != 0) { fts_sanitizer_forget_unwound_frames(); } + /* Release callback-only roots before error-path garbage collection. */ + if (ctx->json_event_ref != LUA_NOREF) { + lua_pushboolean(l, 0); + lua_rawseti(l, LUA_REGISTRYINDEX, ctx->json_event_ref); + } if (ret == 0 && !ctx->jit && ctx->instructions_left <= 0) { lua_settop(l, base); lua_pushliteral(l, "Lua instruction limit exceeded"); diff --git a/plugins/processor_fts/fts_value_helpers.lua b/plugins/processor_fts/fts_value_helpers.lua index 50c6bf10288..626f89829e7 100644 --- a/plugins/processor_fts/fts_value_helpers.lua +++ b/plugins/processor_fts/fts_value_helpers.lua @@ -1,4 +1,25 @@ local fb = package.loaded.fluentbit +--- Parse JSON into a typed value, or return nil and the ordinary parse error. +function fb.try_parse_json(text) + local ok, value = pcall(fb.parse_json, text) + if ok then return value end + return nil, value +end +--- Decode one string field in place; return false and an error without mutating on failure. +function fb.decode_json(container, key) + local source = container:get(key) + if not source or source:type() ~= 'string' then + return false, 'JSON source field must be a string' + end + local value, err = fb.try_parse_json(source:value()) + if not value then return false, err end + if fb._fts_json_integer_error then + err = fb._fts_json_integer_error(container, value) + if err then return false, err end + end + container:set(key, value) + return true +end function fb.keep_keys(map, keys) local keep = {}; for _, k in ipairs(keys) do keep[k] = true end for _, k in ipairs(map:keys()) do if not keep[k] then map:remove(k) end end diff --git a/tests/integration/scenarios/processor_fts/tests/test_fts_records.py b/tests/integration/scenarios/processor_fts/tests/test_fts_records.py index 5e03623b347..69ffd8eafca 100644 --- a/tests/integration/scenarios/processor_fts/tests/test_fts_records.py +++ b/tests/integration/scenarios/processor_fts/tests/test_fts_records.py @@ -72,6 +72,116 @@ def send(service, records): connection.sendall(message) +@pytest.mark.parametrize("placement", ["input", "output"]) +@pytest.mark.parametrize("decoder", ["try_parse_json", "unpack_json"]) +def test_json_embedded_messages(tmp_path, placement, decoder): + """Parse or unpack mixed JSON and plain messages without losing exact integers.""" + code = ''' + local fb = require("fluentbit") + function process(log) + local app, err = fb.try_parse_json(log.body:get_string("message")) + if not app then + assert(type(err) == "string") + return + end + assert(err == nil) + if app:type() ~= "map" then return end + fb.redact(app, {"authorization", "password"}) + log.body = app + end + ''' + if decoder == "unpack_json": + code = ''' + local fb = require("fluentbit") + function process(log) + fb.unpack_json(log, "message") + fb.redact(log.body, {"authorization", "password"}) + end + ''' + records = [ + {"message": json.dumps({"level": "error", "user": "ana", "password": "secret", + "nested": {"authorization": "Bearer token"}})}, + {"message": "plain text line"}, + {"message": '{"level":"info","order_id":9007199254740993}'}, + {"message": '{"order_id":18446744073709551615}'}, + {"message": ' \n\t{"ok":true}'}, + {"message": '{"broken":'}, + {"message": '{"duplicate":1,"duplicate":2}'}, + {"message": '{} trailing'}, + {"message": '18446744073709551616'}, + {"message": ""}, + {"other": "missing message"}, + {"message": "null"}, + {"message": "false"}, + {"message": '[{"password":"secret"}]'}, + {"message": '"JSON string"'}, + ] + expected = [ + {"level": "error", "user": "ana", "password": "[REDACTED]", + "nested": {"authorization": "[REDACTED]"}}, + records[1], + {"level": "info", "order_id": 9007199254740993}, + {"order_id": 18446744073709551615}, + {"ok": True}, + *records[5:], + ] + if decoder == "unpack_json": + records.append({"message": 42}) + expected.append({"message": 42}) + service = start_service(tmp_path, code, placement) + try: + send(service, records) + output = service.wait_for_condition( + lambda: rows(service) if len(rows(service)) >= len(expected) else None, + timeout=45, description="decoded embedded JSON messages") + assert [{key: value for key, value in row.items() if key != "date"} + for row in output] == expected + finally: + service.stop() + + +@pytest.mark.parametrize("placement", ["input", "output"]) +def test_json_fields_decode_in_place(tmp_path, placement): + """Body, nested map, and array fields use the same decoder and keep siblings.""" + code = ''' + local fb = require("fluentbit") + function process(log) + fb.decode_json(log.body, "message") + fb.decode_json(log.body:get("attributes"), "payload") + fb.decode_json(log.body:get("values"), 1) + fb.redact(log.body, {"password"}) + end + ''' + fixtures = [ + ('{"nested":{"password":"secret"}}', {"nested": {"password": "[REDACTED]"}}), + ('{"id":9007199254740993}', {"id": 9007199254740993}), + ('[1,2]', [1, 2]), + ('null', None), + ('false', False), + ('18446744073709551615', 18446744073709551615), + ('"text"', "text"), + ('plain text', 'plain text'), + ('{"broken":', '{"broken":'), + ('{"a":1,"a":2}', '{"a":1,"a":2}'), + (42, 42), + ] + records = [{"message": text, "host": "web", "attributes": {"payload": text, "keep": "s"}, + "values": [text, "sibling"]} for text, _ in fixtures] + expected = [{"message": value, "host": "web", "attributes": {"payload": value, "keep": "s"}, + "values": [value, "sibling"]} for _, value in fixtures] + service = start_service(tmp_path, code, placement) + try: + send(service, records) + output = service.wait_for_condition( + lambda: rows(service) if len(rows(service)) >= len(expected) else None, + timeout=45, description="decoded fields with preserved siblings") + assert [{key: value for key, value in row.items() if key != "date"} + for row in output] == expected + assert "policy failed" not in log_text(service) + finally: + service.stop() + + @pytest.mark.parametrize("placement", ["input", "output"]) @pytest.mark.parametrize("chain", ["direct", "filter", "native", "replacement"]) @pytest.mark.parametrize("error_mode", ["keep", "drop"]) @@ -524,11 +634,12 @@ def test_missing_transport_ids_are_group_local(tmp_path, error_mode): service.stop() -def start_otlp_json_service(tmp_path, processors): +def start_otlp_json_service(tmp_path, processors, placement="input"): + """Start a forward-to-OTLP JSON pipeline with processors at the selected stage.""" source = {"name": "forward", "port": "${FLUENT_BIT_TEST_LISTENER_PORT}", "tag": "stream"} - if processors: - source["processors"] = {"logs": processors} output = {"name": "stdout", "match": "*", "format": "otlp_json"} + if processors: + (source if placement == "input" else output)["processors"] = {"logs": processors} config = tmp_path / "otlp-json.yaml" config.write_text(yaml.safe_dump({"service": {"flush": 1, "grace": 1, "http_server": True, "http_port": "${FLUENT_BIT_HTTP_MONITORING_PORT}"}, @@ -538,6 +649,182 @@ def start_otlp_json_service(tmp_path, processors): return service +@pytest.mark.parametrize("placement", ["input", "output"]) +@pytest.mark.parametrize("invalid_context", [False, True]) +def test_json_fields_decode_otel_attributes(tmp_path, placement, invalid_context): + """Decode actual OTel attribute maps and verify their exported values.""" + code = ''' + local fb = require("fluentbit") + function process(log) + assert(fb.decode_json(log.attributes, "payload")) + assert(fb.decode_json(log.resource.attributes, "labels")) + fb.redact(log.attributes, {"password"}) + assert(log.body:get_string("message") == "body stays unchanged") + end + ''' + if invalid_context: + code = code.replace('function process(log)', '''function process(log) + local context = log.context + local scratch = fb.map() + local function decode() + scratch:set("payload", "18446744073709551615") + assert(fb.decode_json(scratch, "payload")) + assert(scratch:get("payload"):value() == "18446744073709551615") + end + for _, value in ipairs({false, 42, "context", fb.map(), fb.array()}) do + log.context = value + decode() + end + log.context = nil + decode() + log.context = context + ''', 1) + service = start_otlp_json_service(tmp_path, [{"name": "fts", "code": code}], placement) + try: + start, end = event_time(0xFFFFFFFF), event_time(0xFFFFFFFE) + now = event_time(1700000000) + entries = [ + [[start, {"schema": "otlp", "resource_id": 0, "scope_id": 0}], + {"resource": {"attributes": {"labels": '{"region":"east"}', "keep": "r"}}, + "scope": {"name": "json-test"}}], + [[now, {"otlp": {"attributes": { + "payload": '{"id":9007199254740993,"password":"secret"}', "keep": "a"}}}], + {"message": "body stays unchanged"}], + [[end, {}], {}], + ] + with socket.create_connection(("127.0.0.1", service.flb_listener_port), timeout=30) as connection: + connection.sendall(pack(["stream", entries])) + output = service.wait_for_condition(lambda: rows(service), timeout=45, + description="decoded OTel attributes") + resources = [resource for document in output for resource in document["resourceLogs"]] + assert len(resources) == 1 + resource = resources[0] + attrs = {item["key"]: item["value"] for item in resource["resource"]["attributes"]} + assert attrs["keep"] == {"stringValue": "r"} + assert attrs["labels"] == {"kvlistValue": {"values": [ + {"key": "region", "value": {"stringValue": "east"}}]}} + logs = [log for scope in resource["scopeLogs"] for log in scope["logRecords"]] + assert len(logs) == 1 + attrs = {item["key"]: item["value"] for item in logs[0]["attributes"]} + assert attrs["keep"] == {"stringValue": "a"} + payload = {item["key"]: item["value"] + for item in attrs["payload"]["kvlistValue"]["values"]} + assert payload == {"id": {"intValue": "9007199254740993"}, + "password": {"stringValue": "[REDACTED]"}} + assert logs[0]["body"] == {"stringValue": "body stays unchanged"} + assert "policy failed" not in log_text(service) + finally: + service.stop() + + +@pytest.mark.parametrize("placement", ["input", "output"]) +def test_json_otel_unsigned_bounds(tmp_path, placement): + """OTLP decode rejects unsigned overflow before any field is changed.""" + code = ''' + local fb = require("fluentbit") + function process(log) + local allowed = log.body:get_number("allowed") == 1 + local function check(container, key) + local original = container:get(key):value() + local ok, err = fb.decode_json(container, key) + assert(ok == allowed) + if allowed then assert(err == nil) + else + assert(type(err) == "string") + assert(container:get(key):value() == original) + end + end + if log.body:get_string("operation") == "unpack" then + local body = log.body + local ok, err = fb.unpack_json(log, "encoded") + assert(ok == allowed) + if not allowed then assert(type(err) == "string" and log.body == body) end + else + check(log.body, "payload") + check(log.body:get("nested"), "payload") + check(log.body:get("values"), 1) + check(log.attributes, "payload") + check(log.resource.attributes, "payload") + check(log.scope.attributes, "payload") + end + local scratch = fb.map() + scratch:set("payload", "18446744073709551615") + assert(fb.decode_json(scratch, "payload")) + assert(scratch:get("payload"):value() == "18446744073709551615") + log.body:set("checked", true) + end + ''' + + def any_value(value): + """Decode exported AnyValue fixtures without rounding integer strings.""" + if "kvlistValue" in value: + return {item["key"]: any_value(item["value"]) + for item in value["kvlistValue"]["values"]} + if "arrayValue" in value: + return [any_value(item) for item in value["arrayValue"]["values"]] + if "intValue" in value: + return int(value["intValue"]) + if "boolValue" in value: + return value["boolValue"] + return value["stringValue"] + + fixtures = [] + entries = [] + start, end = event_time(0xFFFFFFFF), event_time(0xFFFFFFFE) + now = event_time(1700000000) + for operation in ("decode", "unpack"): + for number in (9223372036854775807, 9223372036854775808, 18446744073709551615): + index = len(fixtures) + allowed = number <= 9223372036854775807 + text = json.dumps({"outer": [{"id": number}]}) + body = {"operation": operation, "allowed": int(allowed), "encoded": text, + "payload": text, "nested": {"payload": text}, "values": [text]} + attrs = {"payload": text, "record_id": str(index)} + entries.extend([ + [[start, {"schema": "otlp", "resource_id": index, "scope_id": index}], + {"resource": {"attributes": dict(attrs)}, + "scope": {"name": "bounds", "attributes": dict(attrs)}}], + [[now, {"otlp": {"attributes": dict(attrs)}}], body], + [[end, {}], {}], + ]) + fixtures.append((operation, allowed, text, body)) + + service = start_otlp_json_service(tmp_path, [{"name": "fts", "code": code}], placement) + try: + with socket.create_connection(("127.0.0.1", service.flb_listener_port), timeout=30) as connection: + connection.sendall(pack(["stream", entries])) + output = service.wait_for_condition( + lambda: rows(service) if sum(len(scope["logRecords"]) + for document in rows(service) for resource in document["resourceLogs"] + for scope in resource["scopeLogs"]) >= len(fixtures) else None, + timeout=45, description="OTLP records with recoverable JSON overflow") + seen = [] + for document in output: + for resource in document["resourceLogs"]: + for scope in resource["scopeLogs"]: + for log in scope["logRecords"]: + attrs = {item["key"]: any_value(item["value"]) for item in log["attributes"]} + index = int(attrs["record_id"]) + seen.append(index) + operation, allowed, text, body = fixtures[index] + value = json.loads(text) if allowed else text + expected = dict(body) + if operation == "decode": + expected.update(payload=value, nested={"payload": value}, values=[value]) + elif allowed: + expected = json.loads(text) + expected["checked"] = True + assert any_value(log["body"]) == expected + for attributes in (log["attributes"], resource["resource"]["attributes"], + scope["scope"]["attributes"]): + decoded = {item["key"]: any_value(item["value"]) for item in attributes} + assert decoded["payload"] == (value if operation == "decode" else text) + assert sorted(seen) == list(range(len(fixtures))) + assert "policy failed" not in log_text(service) + finally: + service.stop() + + def otlp_resources(service): """(resource attributes, record bodies) for every exported resource.""" resources = [] diff --git a/tests/internal/fts.c b/tests/internal/fts.c index 4728c753b49..32cfcb70ea7 100644 --- a/tests/internal/fts.c +++ b/tests/internal/fts.c @@ -431,6 +431,147 @@ static void strict_parsing_and_time(void) 0, "drop", 6, 0, 0, NULL, NULL); } +/* Parsing keeps exact typed values and reports ordinary errors as return values. */ +static void try_parse_json_preserves_types_and_errors(void) +{ + run_policy("local fb=require('fluentbit'); function process(log) " + "local value,err=fb.try_parse_json('18446744073709551615'); " + "assert(value:type()=='uint64' and value:value()=='18446744073709551615'); " + "assert(err==nil); " + "assert(fb.try_parse_json('-9223372036854775808'):value()=='-9223372036854775808'); " + "assert(fb.try_parse_json('null'):type()=='null'); " + "assert(fb.try_parse_json('false'):value()==false); " + "assert(fb.try_parse_json('[1,2]'):len()==2); " + "assert(fb.try_parse_json(' \\n {\"nested\":{\"ok\":true}}')" + ":get('nested'):get('ok'):value()==true); " + "local function invalid(text) local v,e=fb.try_parse_json(text); " + "assert(v==nil and type(e)=='string') end; " + "invalid(nil); invalid(false); invalid(42); invalid({}); invalid(''); " + "invalid('plain text'); invalid('{'); invalid('{} garbage'); " + "invalid('{\"a\":1,\"a\":2}'); invalid('18446744073709551616'); " + "invalid(string.rep(' ',1048577)); " + "invalid(string.rep('[',65)..'0'..string.rep(']',65)); " + "assert(not pcall(fb.parse_json,'plain text')) end", + 0, "drop", 6, 0, 0, NULL, NULL); +} + +/* Unpacking replaces the body only after an object has been decoded successfully. */ +static void unpack_json_replaces_only_object_bodies(void) +{ + run_policy("local fb=require('fluentbit'); function process(log) " + "local record={body=fb.map()}; local original=record.body; " + "local ok,err=fb.unpack_json(record); " + "assert(ok==false and type(err)=='string' and record.body==original); " + "for _,text in ipairs({'plain text','{','[]','null','false','42'}) do " + "record.body:set('message',text); ok,err=fb.unpack_json(record); " + "assert(ok==false and type(err)=='string' and record.body==original); " + "assert(record.body:get_string('message')==text) end; " + "record.body:set('message',42); assert(not fb.unpack_json(record)); " + "assert(record.body==original); " + "record.body:set('payload','{\"id\":18446744073709551615}'); " + "ok,err=fb.unpack_json(record,'payload'); assert(ok==true and err==nil); " + "assert(record.body:get('id'):value()=='18446744073709551615'); " + "record.body:set('message','{}'); assert(fb.unpack_json(record)); " + "assert(record.body:len()==0) end", + 0, "drop", 6, 0, 0, NULL, NULL); +} + +/* Field decoding preserves siblings, scalar types, and failed input strings. */ +static void decode_json_fields_preserve_siblings_and_types(void) +{ + run_policy("local fb=require('fluentbit'); function process(log) " + "local map=fb.map(); map:set('sibling','keep'); " + "local function decode(text,kind) map:set('payload',text); " + "local ok,err=fb.decode_json(map,'payload'); assert(ok==true and err==nil); " + "assert(map:get('payload'):type()==kind and map:get_string('sibling')=='keep'); " + "return map:get('payload') end; " + "assert(decode('{\"id\":18446744073709551615}','map')" + ":get('id'):value()=='18446744073709551615'); " + "assert(decode('[1,2]','array'):len()==2); decode('null','null'); " + "assert(decode('false','bool'):value()==false); " + "assert(decode('\"text\"','string'):value()=='text'); " + "assert(decode('9007199254740993','uint64'):value()=='9007199254740993'); " + "for _,text in ipairs({'plain text','{','{} trailing','{\"a\":1,\"a\":2}'}) do " + "map:set('payload',text); local ok,err=fb.decode_json(map,'payload'); " + "assert(ok==false and type(err)=='string' and map:get_string('payload')==text); " + "assert(map:get_string('sibling')=='keep') end; " + "assert(not fb.decode_json(map,'missing')); " + "map:set('payload',42); assert(not fb.decode_json(map,'payload')); " + "assert(map:get_number('payload')==42); " + "local array=fb.array(); array:append('false'); array:append('null'); " + "assert(fb.decode_json(array,1) and array:get(1):value()==false); " + "assert(fb.decode_json(array,2) and array:get(2):type()=='null'); " + "assert(not pcall(fb.decode_json,map,1)); " + "assert(not pcall(fb.decode_json,nil,'payload')) end", + 0, "drop", 6, 0, 0, NULL, NULL); +} + +/* Both JSON mutation helpers retain terminal instruction-budget semantics. */ +static void json_helpers_cannot_swallow_instruction_exhaustion(void) +{ + const char *decode; + const char *unpack; + + decode = "local fb=require('fluentbit'); function process(log) " + "log.body:set('payload','{\"items\":['..string.rep('0,',30000)..'0]}'); " + "fb.decode_json(log.body,'payload') end"; + unpack = "local fb=require('fluentbit'); function process(log) " + "log.body:set('payload','{\"items\":['..string.rep('0,',30000)..'0]}'); " + "fb.unpack_json(log,'payload') end"; + run_policy(decode, 0, "drop", 0, 0, 0, NULL, NULL); + run_policy(unpack, 0, "drop", 0, 0, 0, NULL, NULL); + run_policy(decode, 0, "keep", 6, 0, 0, NULL, NULL); + run_policy(unpack, 0, "keep", 6, 0, 0, NULL, NULL); +} + +/* Reject OTLP integer overflow without restricting generic parser or scratch values. */ +static void json_helpers_reject_otel_unsigned_overflow(void) +{ + run_policy("local fb=require('fluentbit'); function process(log) " + "local function reject(container,text) container:set('payload',text); " + "local ok,err=fb.decode_json(container,'payload'); " + "assert(ok==false and type(err)=='string'); " + "assert(container:get_string('payload')==text) end; " + "for _,container in ipairs({log.body,log.attributes," + "log.resource.attributes,log.scope.attributes,log.body:ensure_map('nested')}) do " + "reject(container,'{\"nested\":[{\"id\":18446744073709551615}]}'); " + "reject(container,'[{\"id\":9223372036854775808}]'); " + "reject(container,'18446744073709551615'); " + "container:set('payload','{\"id\":9223372036854775807}'); " + "assert(fb.decode_json(container,'payload')); " + "assert(container:get('payload'):get('id'):value()=='9223372036854775807') end; " + "local array=fb.array(); array:append('18446744073709551615'); " + "log.body:set('array',array); " + "assert(not fb.decode_json(log.body:get('array'),1)); " + "assert(log.body:get('array'):get(1):value()=='18446744073709551615'); " + "local scratch=fb.map(); scratch:set('payload','18446744073709551615'); " + "assert(fb.decode_json(scratch,'payload')); " + "assert(scratch:get('payload'):value()=='18446744073709551615'); " + "assert(fb.try_parse_json('18446744073709551615'):value()=='18446744073709551615'); " + "log.body=log.body:copy(); reject(log.body,'9223372036854775808'); " + "log.body:set('message','{\"id\":9223372036854775808}'); " + "local body=log.body; local ok,err=fb.unpack_json(log); " + "assert(ok==false and type(err)=='string' and log.body==body) end", + 3, "drop", 6, 2, 0, NULL, NULL); +} + +/* A scratch decode must not read a temporarily invalid context as a Lua table. */ +static void json_helpers_ignore_non_table_contexts(void) +{ + run_policy("local fb=require('fluentbit'); function process(log) " + "local context=log.context; local scratch=fb.map(); " + "local function decode() scratch:set('payload','18446744073709551615'); " + "assert(fb.decode_json(scratch,'payload')); " + "assert(scratch:get('payload'):value()=='18446744073709551615') end; " + "for _,value in ipairs({false,42,'context',fb.map(),fb.array()}) do " + "log.context=value; decode() end; " + "log.context=nil; decode(); log.context=context; " + "log.resource.attributes:set('payload','9223372036854775808'); " + "assert(not fb.decode_json(log.resource.attributes,'payload')); " + "assert(log.resource.attributes:get_string('payload')=='9223372036854775808') end", + 3, "drop", 6, 2, 0, NULL, NULL); +} + static void output_growth_fails_closed(void) { run_policy("function process(log) log.body:set('grow', string.rep('x',300000)) end", @@ -2383,6 +2524,12 @@ TEST_LIST = { {"vm_memory_exhaustion_drops", vm_memory_exhaustion_drops}, {"native_memory_exhaustion_drops", native_memory_exhaustion_drops}, {"strict_parsing_and_time", strict_parsing_and_time}, + {"try_parse_json_types_and_errors", try_parse_json_preserves_types_and_errors}, + {"unpack_json_object_bodies", unpack_json_replaces_only_object_bodies}, + {"decode_json_fields", decode_json_fields_preserve_siblings_and_types}, + {"json_helpers_instruction_exhaustion", json_helpers_cannot_swallow_instruction_exhaustion}, + {"json_helpers_otel_unsigned_overflow", json_helpers_reject_otel_unsigned_overflow}, + {"json_helpers_non_table_contexts", json_helpers_ignore_non_table_contexts}, {"output_growth_fails_closed", output_growth_fails_closed}, {"output_growth_keep_restores_whole_chunk", output_growth_keep_restores_whole_chunk}, {"initialization_failures", initialization_failures},