diff --git a/plugins/in_kubernetes_events/kubernetes_events.c b/plugins/in_kubernetes_events/kubernetes_events.c index 3d996806b54..291d90d020f 100644 --- a/plugins/in_kubernetes_events/kubernetes_events.c +++ b/plugins/in_kubernetes_events/kubernetes_events.c @@ -21,6 +21,8 @@ #include #include #include +#include +#include #include #include @@ -179,6 +181,7 @@ 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; @@ -186,13 +189,23 @@ static msgpack_object *record_get_field_ptr(msgpack_object *obj, const char *fie 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) { v = &obj->via.map.ptr[i].val; return v; } @@ -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; @@ -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) { @@ -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; } @@ -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) { @@ -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; @@ -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; } @@ -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; } @@ -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; } @@ -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); @@ -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; } @@ -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; } @@ -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; @@ -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) { @@ -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"); if (meta == NULL) { flb_plg_error(ctx->ins, "unable to find metadata to save event"); return -1; @@ -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; } diff --git a/plugins/in_kubernetes_events/kubernetes_events_conf.c b/plugins/in_kubernetes_events/kubernetes_events_conf.c index 26ccc850ad3..315a07a378c 100644 --- a/plugins/in_kubernetes_events/kubernetes_events_conf.c +++ b/plugins/in_kubernetes_events/kubernetes_events_conf.c @@ -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 diff --git a/tests/integration/scenarios/in_kubernetes_events/tests/test_in_kubernetes_events_001.py b/tests/integration/scenarios/in_kubernetes_events/tests/test_in_kubernetes_events_001.py index 3a258cf5fde..8d9a8fc0b3e 100644 --- a/tests/integration/scenarios/in_kubernetes_events/tests/test_in_kubernetes_events_001.py +++ b/tests/integration/scenarios/in_kubernetes_events/tests/test_in_kubernetes_events_001.py @@ -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): @@ -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): @@ -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) @@ -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) @@ -117,6 +127,7 @@ 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( @@ -124,7 +135,7 @@ def _write_config(tmp_path, kube_api_port): "[SERVICE]", " Flush 1", " Grace 1", - " Log_Level info", + " Log_Level debug", " HTTP_Server On", " HTTP_Port ${FLUENT_BIT_HTTP_MONITORING_PORT}", "", @@ -132,6 +143,7 @@ def _write_config(tmp_path, kube_api_port): " 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", @@ -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