Skip to content

in_kubernetes_events: fix unsafe MessagePack field parsing #12222

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
edsiper merged 8 commits into from
Aug 5, 2026
+117 −45
Merged

in_kubernetes_events: fix unsafe MessagePack field parsing #12222

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
d5ceb22
in_kubernetes_events: fix OOB reads from non-NUL-terminated msgpack s…
zanarellidev Jul 31, 2026
f8c40d6
in_kubernetes_events: tighten timestamp and uint64 parsing validation
zanarellidev Jul 31, 2026
c975573
in_kubernetes_events: reject signed uint64 strings in record_get_fiel…
zanarellidev Jul 31, 2026
77d2063
in_kubernetes_events: complete bounded field lookup
edsiper Aug 4, 2026
87bbdc6
in_kubernetes_events: finalize database statements
edsiper Aug 4, 2026
2c6b207
tests: integration: cover kubernetes events database
edsiper Aug 4, 2026
e18cb1f
in_kubernetes_events: reject missing string fields
edsiper Aug 4, 2026
a6bf1db
tests: integration: cover missing event fields
edsiper Aug 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
125 changes: 88 additions & 37 deletions plugins/in_kubernetes_events/kubernetes_events.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,8 @@
#include <sys/types.h>
#include <sys/stat.h>
#include <inttypes.h>
#include <errno.h>
#include <ctype.h>

#include <fluent-bit/flb_input_plugin.h>
#include <fluent-bit/flb_network.h>
Expand Down Expand Up @@ -179,20 +181,31 @@ static int refresh_token_if_needed(struct k8s_events *ctx)
static msgpack_object *record_get_field_ptr(msgpack_object *obj, const char *fieldname)
{
int i;
size_t fieldname_len;
msgpack_object *k;
msgpack_object *v;

if (obj->type != MSGPACK_OBJECT_MAP) {
return NULL;
}

fieldname_len = strlen(fieldname);

for (i = 0; i < obj->via.map.size; i++) {
k = &obj->via.map.ptr[i].key;
if (k->type != MSGPACK_OBJECT_STR) {
continue;
}

if (strncmp(k->via.str.ptr, fieldname, strlen(fieldname)) == 0) {
/*
* msgpack strings are not NUL terminated: k->via.str.ptr points
* directly into the decode buffer for exactly k->via.str.size
* bytes. Require an exact length match before comparing so we
* never read past that boundary, and so a key that merely shares
* a prefix with fieldname cannot match.
*/
if ((size_t) k->via.str.size == fieldname_len &&
strncmp(k->via.str.ptr, fieldname, fieldname_len) == 0) {
Comment thread
edsiper marked this conversation as resolved.
v = &obj->via.map.ptr[i].val;
return v;
}
Expand All @@ -204,6 +217,8 @@ static int record_get_field_sds(msgpack_object *obj, const char *fieldname, flb_
{
msgpack_object *v;

*val = NULL;

v = record_get_field_ptr(obj, fieldname);
if (v == NULL) {
return 0;
Expand All @@ -213,13 +228,19 @@ static int record_get_field_sds(msgpack_object *obj, const char *fieldname, flb_
}

*val = flb_sds_create_len(v->via.str.ptr, v->via.str.size);
if (*val == NULL) {
return -1;
}

return 0;
}

static int record_get_field_time(msgpack_object *obj, const char *fieldname, struct flb_time *val)
{
char *end;
msgpack_object *v;
struct flb_tm tm = { 0 };
char buf[64];

v = record_get_field_ptr(obj, fieldname);
if (v == NULL) {
Expand All @@ -229,7 +250,20 @@ static int record_get_field_time(msgpack_object *obj, const char *fieldname, str
return -1;
}

if (flb_strptime(v->via.str.ptr, "%Y-%m-%dT%H:%M:%SZ", &tm) == NULL) {
/*
* msgpack strings are not NUL terminated: v->via.str.ptr points
* directly into the decode buffer for exactly v->via.str.size bytes.
* Copy it into a bounded, NUL-terminated stack buffer before handing
* it to flb_strptime(), instead of scanning the raw buffer directly.
*/
if (v->via.str.size == 0 || v->via.str.size >= sizeof(buf)) {
return -2;
}
memcpy(buf, v->via.str.ptr, v->via.str.size);
buf[v->via.str.size] = '\0';

end = flb_strptime(buf, "%Y-%m-%dT%H:%M:%SZ", &tm);
if (end == NULL || *end != '\0') {
return -2;
}

Expand All @@ -242,7 +276,9 @@ static int record_get_field_time(msgpack_object *obj, const char *fieldname, str
static int record_get_field_uint64(msgpack_object *obj, const char *fieldname, uint64_t *val)
{
msgpack_object *v;
char buf[32];
char *end;
size_t len;

v = record_get_field_ptr(obj, fieldname);
if (v == NULL) {
Expand All @@ -251,8 +287,35 @@ static int record_get_field_uint64(msgpack_object *obj, const char *fieldname, u

/* attempt to parse string as number... */
if (v->type == MSGPACK_OBJECT_STR) {
*val = strtoul(v->via.str.ptr, &end, 10);
if (end == NULL || (end < v->via.str.ptr + v->via.str.size)) {
/*
* msgpack strings are not NUL terminated: v->via.str.ptr points
* directly into the decode buffer for exactly v->via.str.size
* bytes. Copy it into a bounded, NUL-terminated stack buffer
* before calling strtoul() on it, instead of scanning the raw
* buffer directly (no valid uint64 needs more than 20 digits).
*/
len = v->via.str.size;
if (len == 0 || len > sizeof(buf) - 1) {
return -1;
}
memcpy(buf, v->via.str.ptr, len);
buf[len] = '\0';

/*
* strtoull() itself accepts a leading '+'/'-' and skips leading
* whitespace, which would let a value like "-5" silently wrap
* around into a huge positive number instead of being rejected.
* Kubernetes resourceVersion (the only caller) is always a plain,
* unsigned, digits-only decimal string, so require that directly
* before parsing.
*/
if (!isdigit((unsigned char) buf[0])) {
return -1;
}

errno = 0;
*val = strtoull(buf, &end, 10);
if (errno == ERANGE || end == buf || *end != '\0') {
return -1;
}
return 0;
Expand All @@ -262,8 +325,7 @@ static int record_get_field_uint64(msgpack_object *obj, const char *fieldname, u
return 0;
}
if (v->type == MSGPACK_OBJECT_NEGATIVE_INTEGER) {
*val = (uint64_t)v->via.i64;
return 0;
return -1;
}
return -1;
}
Expand All @@ -277,12 +339,12 @@ static int item_get_timestamp(msgpack_object *obj, struct flb_time *event_time)
* NULL while having metadata.creationTimestamp set.
*/
ret = record_get_field_time(obj, "lastTimestamp", event_time);
if (ret != -1) {
if (ret == 0) {
return FLB_TRUE;
}

ret = record_get_field_time(obj, "firstTimestamp", event_time);
if (ret != -1) {
if (ret == 0) {
return FLB_TRUE;
}

Expand All @@ -292,7 +354,7 @@ static int item_get_timestamp(msgpack_object *obj, struct flb_time *event_time)
}

ret = record_get_field_time(metadata, "creationTimestamp", event_time);
if (ret != -1) {
if (ret == 0) {
return FLB_TRUE;
}

Expand All @@ -305,7 +367,7 @@ static bool check_event_is_filtered(struct k8s_events *ctx, msgpack_object *obj,
int ret;
uint64_t outdated;
msgpack_object *metadata;
flb_sds_t uid;
flb_sds_t uid = NULL;
uint64_t resource_version;

outdated = cfl_time_now() - ((uint64_t) ctx->retention_time * 1000000000ULL);
Expand All @@ -328,8 +390,8 @@ static bool check_event_is_filtered(struct k8s_events *ctx, msgpack_object *obj,
}

ret = record_get_field_sds(metadata, "uid", &uid);
if (ret == -1) {
flb_plg_error(ctx->ins, "Cannot get resourceVersion for item in response");
if (ret == -1 || uid == NULL) {
flb_plg_error(ctx->ins, "Cannot get uid for item in response");
return FLB_FALSE;
}

Expand Down Expand Up @@ -470,8 +532,9 @@ static int process_watched_event(struct k8s_events *ctx, char *buf_data, size_t
}

ret = record_get_field_sds(&root, "type", &event_type);
if (ret == -1) {
if (ret == -1 || event_type == NULL) {
flb_plg_warn(ctx->ins, "Streamed Event 'type' not found");
ret = -1;
goto msg_error;
}

Expand Down Expand Up @@ -511,7 +574,6 @@ static int process_event_list(struct k8s_events *ctx, char *in_data, size_t in_s
size_t off = 0;
msgpack_unpacked result;
msgpack_object root;
msgpack_object k;
msgpack_object *items = NULL;
msgpack_object *item = NULL;
msgpack_object *metadata = NULL;
Expand Down Expand Up @@ -540,27 +602,16 @@ static int process_event_list(struct k8s_events *ctx, char *in_data, size_t in_s
/* Traverse the EventList for the metadata (for the continue token) and the items.
* https://kubernetes.io/docs/reference/kubernetes-api/cluster-resources/event-v1/#EventList
*/
for (i = 0; i < root.via.map.size; i++) {
k = root.via.map.ptr[i].key;
if (k.type != MSGPACK_OBJECT_STR) {
continue;
}

if (strncmp(k.via.str.ptr, "items", 5) == 0) {
items = &root.via.map.ptr[i].val;
if (items->type != MSGPACK_OBJECT_ARRAY) {
flb_plg_error(ctx->ins, "Cannot unpack items");
goto msg_error;
}
}
items = record_get_field_ptr(&root, "items");
if (items != NULL && items->type != MSGPACK_OBJECT_ARRAY) {
flb_plg_error(ctx->ins, "Cannot unpack items");
goto msg_error;
}

if (strncmp(k.via.str.ptr, "metadata", 8) == 0) {
metadata = &root.via.map.ptr[i].val;
if (metadata->type != MSGPACK_OBJECT_MAP) {
flb_plg_error(ctx->ins, "Cannot unpack metadata");
goto msg_error;
}
}
metadata = record_get_field_ptr(&root, "metadata");
if (metadata != NULL && metadata->type != MSGPACK_OBJECT_MAP) {
flb_plg_error(ctx->ins, "Cannot unpack metadata");
goto msg_error;
}

if (items == NULL) {
Expand Down Expand Up @@ -695,10 +746,10 @@ static int k8s_events_sql_insert_event(struct k8s_events *ctx, msgpack_object *i
uint64_t resource_version;
struct flb_time last;
msgpack_object *meta;
flb_sds_t uid;
flb_sds_t uid = NULL;


meta = record_get_field_ptr(item, "meta");
meta = record_get_field_ptr(item, "metadata");
Comment thread
coderabbitai[bot] marked this conversation as resolved.
if (meta == NULL) {
flb_plg_error(ctx->ins, "unable to find metadata to save event");
return -1;
Expand All @@ -711,7 +762,7 @@ static int k8s_events_sql_insert_event(struct k8s_events *ctx, msgpack_object *i
}

ret = record_get_field_sds(meta, "uid", &uid);
if (ret == -1) {
if (ret == -1 || uid == NULL) {
flb_plg_error(ctx->ins, "unable to find uid in metadata to save event");
return -1;
}
Expand Down
3 changes: 3 additions & 0 deletions plugins/in_kubernetes_events/kubernetes_events_conf.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -362,6 +362,9 @@ void k8s_events_conf_destroy(struct k8s_events *ctx)

#ifdef FLB_HAVE_SQLDB
if (ctx->db) {
sqlite3_finalize(ctx->stmt_get_kubernetes_event_exists_by_uid);
sqlite3_finalize(ctx->stmt_insert_kubernetes_event);
sqlite3_finalize(ctx->stmt_delete_old_kubernetes_events);
flb_kubernetes_event_db_close(ctx->db);
}
#endif
Expand Down
34 changes: 26 additions & 8 deletions tests/integration/scenarios/in_kubernetes_events/tests/test_in_kubernetes_events_001.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,15 +13,17 @@
RECOVERED_EVENT_UID = "post-recovery-event-uid"


def _event(resource_version, uid):
def _event(resource_version, uid=None):
timestamp = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
return {
event = {
"metadata": {
"creationTimestamp": timestamp,
"resourceVersion": str(resource_version),
"uid": uid,
}
}
if uid is not None:
event["metadata"]["uid"] = uid
return event


class _KubeApiServer(http.server.ThreadingHTTPServer):
Expand All @@ -36,6 +38,7 @@ def __init__(self, server_address, handler_class):
self.watch_paths = []
self.event = _event(2, EVENT_UID)
self.recovered_event = _event(3, RECOVERED_EVENT_UID)
self.uidless_event = _event(4)


class _KubeApiHandler(http.server.BaseHTTPRequestHandler):
Expand All @@ -54,10 +57,13 @@ def do_GET(self):
self.end_headers()
if watch_request <= 2:
event = self.server.event
type_key = "type"
if watch_request == 1:
type_key = "typeExtra"
if watch_request == 2:
event = self.server.recovered_event
payload = (
json.dumps({"type": "ADDED", "object": event}) + "\n"
json.dumps({type_key: "ADDED", "object": event}) + "\n"
).encode("utf-8")
self.wfile.write(f"{len(payload):x}\r\n".encode("ascii"))
self.wfile.write(payload)
Expand Down Expand Up @@ -85,7 +91,11 @@ def do_GET(self):
"kind": "EventList",
"apiVersion": "v1",
"metadata": {"resourceVersion": str(min(list_request, 2))},
"items": [self.server.event] if list_request > 1 else [],
"items": (
[self.server.event, self.server.uidless_event]
if list_request > 1
else []
),
}
).encode("utf-8")
self.send_response(200)
Expand Down Expand Up @@ -117,21 +127,23 @@ def _run_kube_api_server():
def _write_config(tmp_path, kube_api_port):
token_file = tmp_path / "token"
token_file.write_text("test-token", encoding="utf-8")
database_file = tmp_path / "kubernetes-events.db"
config_file = tmp_path / "kubernetes_events_watch_timeout.conf"
config_file.write_text(
"\n".join(
[
"[SERVICE]",
" Flush 1",
" Grace 1",
" Log_Level info",
" Log_Level debug",
" HTTP_Server On",
" HTTP_Port ${FLUENT_BIT_HTTP_MONITORING_PORT}",
"",
"[INPUT]",
" Name kubernetes_events",
f" Kube_URL http://127.0.0.1:{kube_api_port}",
f" Kube_Token_File {token_file}",
f" Db {database_file}",
" tls Off",
" Interval_Sec 5",
" Interval_NSec 0",
Expand Down Expand Up @@ -173,5 +185,11 @@ def test_kubernetes_events_reconnects_stalled_watch(tmp_path):
assert kube_api_server.list_requests >= 2
assert kube_api_server.watch_requests >= 2
assert all("timeoutSeconds=1" in path for path in kube_api_server.watch_paths)
assert log_text.count(EVENT_UID) == 1
assert log_text.count(RECOVERED_EVENT_UID) == 1
assert log_text.count(f'"uid"=>"{EVENT_UID}"') == 1
assert log_text.count(f'"uid"=>"{RECOVERED_EVENT_UID}"') == 1
assert "Streamed Event 'type' not found" in log_text
assert "Cannot get uid for item in response" in log_text
assert "unable to find uid in metadata to save event" in log_text
assert "unable to find metadata to save event" not in log_text
assert f"inserted k8s event: uid={EVENT_UID}" in log_text
assert f"inserted k8s event: uid={RECOVERED_EVENT_UID}" in log_text
Comment on lines +188 to +195

@coderabbitai coderabbitai Bot Aug 4, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
# Description: Inspect the EventList item-extraction loop and save-event call flow in kubernetes_events.c
ast-grep outline plugins/in_kubernetes_events/kubernetes_events.c --items all --match 'items|uid|save'
rg -n -B5 -A25 'Cannot get uid for item in response' plugins/in_kubernetes_events/kubernetes_events.c
rg -n -B15 -A5 'unable to find uid in metadata to save event' plugins/in_kubernetes_events/kubernetes_events.c

Repository: fluent/fluent-bit

Length of output: 2135

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "== file size =="
wc -l plugins/in_kubernetes_events/kubernetes_events.c tests/integration/scenarios/in_kubernetes_events/tests/test_in_kubernetes_events_001.py

echo "== functions/sections outline grep =="
rg -n 'static|int |flb_plg_error|static int.*event|save|list|items|metadata|uid|EventList' plugins/in_kubernetes_events/kubernetes_events.c \
  | head -n 220

echo "== test relevant sections =="
sed -n '1,130p' tests/integration/scenarios/in_kubernetes_events/tests/test_in_kubernetes_events_001.py
echo "== assertions =="
sed -n '150,210p' tests/integration/scenarios/in_kubernetes_events/tests/test_in_kubernetes_events_001.py

echo "== k8s event extraction/save function sections =="
sed -n '320,430p' plugins/in_kubernetes_events/kubernetes_events.c
sed -n '680,820p' plugins/in_kubernetes_events/kubernetes_events.c

echo "== callers of extraction/save =="
rg -n -B8 -A12 'extract|EventList|unable to find metadata to save event|unable to find uid in metadata to save event|save.*event|kubernetes_event_save|save_event' plugins/in_kubernetes_events/kubernetes_events.c

Repository: fluent/fluent-bit

Length of output: 26413

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "== process_event_list full implementation =="
sed -n '565,670p' plugins/in_kubernetes_events/kubernetes_events.c

echo "== remaining extraction helper usages around process_event_list =="
rg -n -B3 -A8 'process_event_object|check_event_is_filtered|last_resource_version|items->via.array.size' plugins/in_kubernetes_events/kubernetes_events.c

echo "== remaining process_event_object implementation =="
sed -n '441,510p' plugins/in_kubernetes_events/kubernetes_events.c

Repository: fluent/fluent-bit

Length of output: 9066

No change needed. process_event_object() does not check metadata.uid, so the UID-less list item does not emit "unable to find uid in metadata to save event" or reach SQLite insert. This assertion will fail unless the extraction check is skipped or the save path is adjusted.

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In
`@tests/integration/scenarios/in_kubernetes_events/tests/test_in_kubernetes_events_001.py`
around lines 188 - 195, Update the assertions around process_event_object() to
match its current behavior: do not expect a UID-less list item to emit "unable
to find uid in metadata to save event" or reach SQLite insertion. Remove or
revise only the conflicting assertion while preserving the existing checks for
the other event-processing messages and successful inserts.

Loading