decode_opentelemetry: bound wire decoding and attribute nesting

src/ctr_decode_opentelemetry.c

@@ -17,13 +17,15 @@
  *  limitations under the License.
  */
 
+#include "ctr_protobuf.h"
 #include <ctraces/ctraces.h>
+#include <ctraces/ctr_variant_utils.h>
 #include <cfl/cfl_array.h>
 #include <fluent-otel-proto/fluent-otel.h>
 
 static int convert_any_value(struct opentelemetry_decode_value *ctr_val,
                              opentelemetry_decode_value_type value_type, char *key,
-                             Opentelemetry__Proto__Common__V1__AnyValue *val);
+                             Opentelemetry__Proto__Common__V1__AnyValue *val, size_t depth);
 
 static int convert_string_value(struct opentelemetry_decode_value *ctr_val,
                                 opentelemetry_decode_value_type value_type,
@@ -158,14 +160,15 @@ static int convert_double_value(struct opentelemetry_decode_value *ctr_val,
 
 static int convert_array_value(struct opentelemetry_decode_value *ctr_val,
                                opentelemetry_decode_value_type value_type,
-                               char *key, Opentelemetry__Proto__Common__V1__ArrayValue *otel_arr)
+                               char *key, Opentelemetry__Proto__Common__V1__ArrayValue *otel_arr,
+                               size_t depth)
 {
     int array_index;
     int result;
     struct opentelemetry_decode_value *ctr_arr_val;
     Opentelemetry__Proto__Common__V1__AnyValue *val;
 
-    if (otel_arr == NULL) {
+    if (depth >= CFL_VARIANT_UTILS_MAXIMUM_NESTING_DEPTH || otel_arr == NULL) {
         return -1;
     }
     if (otel_arr->n_values > 0 && otel_arr->values == NULL) {
@@ -193,7 +196,7 @@ static int convert_array_value(struct opentelemetry_decode_value *ctr_val,
             /* skip malformed entry rather than failing the whole array */
             continue;
         }
-        result = convert_any_value(ctr_arr_val, CTR_OPENTELEMETRY_TYPE_ARRAY, NULL, val);
+        result = convert_any_value(ctr_arr_val, CTR_OPENTELEMETRY_TYPE_ARRAY, NULL, val, depth + 1);
     }
 
     if (result < 0) {
@@ -230,14 +233,15 @@ static int convert_array_value(struct opentelemetry_decode_value *ctr_val,
 
 static int convert_kvlist_value(struct opentelemetry_decode_value *ctr_val,
                                 opentelemetry_decode_value_type value_type,
-                                char *key, Opentelemetry__Proto__Common__V1__KeyValueList *otel_kvlist)
+                                char *key, Opentelemetry__Proto__Common__V1__KeyValueList *otel_kvlist,
+                                size_t depth)
 {
     int kvlist_index;
     int result;
     struct opentelemetry_decode_value *ctr_kvlist_val;
     Opentelemetry__Proto__Common__V1__KeyValue *kv;
 
-    if (otel_kvlist == NULL) {
+    if (depth >= CFL_VARIANT_UTILS_MAXIMUM_NESTING_DEPTH || otel_kvlist == NULL) {
         return -1;
     }
     if (otel_kvlist->n_values > 0 && otel_kvlist->values == NULL) {
@@ -267,7 +271,8 @@ static int convert_kvlist_value(struct opentelemetry_decode_value *ctr_val,
         if (kv == NULL || kv->key == NULL || kv->value == NULL) {
             continue;
         }
-        result = convert_any_value(ctr_kvlist_val, CTR_OPENTELEMETRY_TYPE_KVLIST, kv->key, kv->value);
+        result = convert_any_value(ctr_kvlist_val, CTR_OPENTELEMETRY_TYPE_KVLIST,
+                                   kv->key, kv->value, depth + 1);
     }
 
     if (result < 0){
@@ -335,7 +340,7 @@ static int convert_bytes_value(struct opentelemetry_decode_value *ctr_val,
 
 static int convert_any_value(struct opentelemetry_decode_value *ctr_val,
                              opentelemetry_decode_value_type value_type, char *key,
-                             Opentelemetry__Proto__Common__V1__AnyValue *val)
+                             Opentelemetry__Proto__Common__V1__AnyValue *val, size_t depth)
 {
     int result;
 
@@ -371,11 +376,11 @@ static int convert_any_value(struct opentelemetry_decode_value *ctr_val,
             break;
 
         case OPENTELEMETRY__PROTO__COMMON__V1__ANY_VALUE__VALUE_ARRAY_VALUE:
-            result = convert_array_value(ctr_val, value_type, key, val->array_value);
+            result = convert_array_value(ctr_val, value_type, key, val->array_value, depth);
             break;
 
         case OPENTELEMETRY__PROTO__COMMON__V1__ANY_VALUE__VALUE_KVLIST_VALUE:
-            result = convert_kvlist_value(ctr_val, value_type, key, val->kvlist_value);
+            result = convert_kvlist_value(ctr_val, value_type, key, val->kvlist_value, depth);
             break;
 
         case OPENTELEMETRY__PROTO__COMMON__V1__ANY_VALUE__VALUE_BYTES_VALUE:
@@ -442,7 +447,7 @@ static struct ctrace_attributes *convert_otel_attrs(size_t n_attributes,
 
         result = convert_any_value(ctr_decoded_attributes,
                                        CTR_OPENTELEMETRY_TYPE_ATTRIBUTE,
-                                       key, val);
+                                       key, val, 1);
     }
 
     if (result < 0) {
@@ -642,6 +647,12 @@ int ctr_decode_opentelemetry_create(struct ctrace **out_ctr,
         return CTR_DECODE_OPENTELEMETRY_INSUFFICIENT_DATA;
     }
 
+    if (ctr_protobuf_validate(
+            &opentelemetry__proto__collector__trace__v1__export_trace_service_request__descriptor,
+            &in_buf[*offset], in_size - *offset) != 0) {
+        return CTR_DECODE_OPENTELEMETRY_CORRUPTED_DATA;
+    }
+
     service_request = opentelemetry__proto__collector__trace__v1__export_trace_service_request__unpack(NULL,
                                                                                                       in_size - *offset,
                                                                                                       (unsigned char *) &in_buf[*offset]);

src/ctr_protobuf.h

@@ -0,0 +1,139 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2015-2026 The Fluent Bit Authors
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+#ifndef CTR_PROTOBUF_H
+#define CTR_PROTOBUF_H
+
+#include <fluent-otel-proto/fluent-otel.h>
+
+/* Budget includes export/resource/scope wrappers and scalar AnyValue messages. */
+#define CTR_PROTOBUF_MAX_DEPTH 100
+
+struct ctr_protobuf_frame {
+    const ProtobufCMessageDescriptor *descriptor;
+    const unsigned char *cursor;
+    const unsigned char *end;
+};
+
+static int ctr_protobuf_read_varint(struct ctr_protobuf_frame *frame, uint64_t *value)
+{
+    unsigned int shift;
+    unsigned char byte;
+
+    *value = 0;
+    for (shift = 0; shift < 64; shift += 7) {
+        if (frame->cursor == frame->end) {
+            return -1;
+        }
+        byte = *frame->cursor++;
+        if (shift == 63 && byte > 1) {
+            return -1;
+        }
+        *value |= (uint64_t) (byte & 0x7f) << shift;
+        if ((byte & 0x80) == 0) {
+            return 0;
+        }
+    }
+    return -1;
+}
+
+/*
+ * Inspect known message fields before protobuf-c recursively allocates them.
+ * Strings, bytes, packed scalars and unknown fields are opaque. The explicit
+ * stack bounds our own stack usage as well as the subsequent unpack operation.
+ */
+static int ctr_protobuf_validate(const ProtobufCMessageDescriptor *descriptor,
+                                    const void *data, size_t size)
+{
+    struct ctr_protobuf_frame frames[CTR_PROTOBUF_MAX_DEPTH];
+    struct ctr_protobuf_frame *frame;
+    const ProtobufCFieldDescriptor *field;
+    const unsigned char *message;
+    uint64_t tag;
+    uint64_t length;
+    size_t depth;
+
+    if (descriptor == NULL || (data == NULL && size != 0)) {
+        return -1;
+    }
+    if (size == 0) {
+        return 0;
+    }
+
+    depth = 1;
+    frames[0].descriptor = descriptor;
+    frames[0].cursor = data;
+    frames[0].end = frames[0].cursor + size;
+
+    while (depth > 0) {
+        frame = &frames[depth - 1];
+        if (frame->cursor == frame->end) {
+            depth--;
+            continue;
+        }
+        if (ctr_protobuf_read_varint(frame, &tag) != 0 || tag >> 3 == 0 || tag >> 3 > 0x1fffffff) {
+            return -1;
+        }
+
+        switch (tag & 7) {
+        case 0:
+            if (ctr_protobuf_read_varint(frame, &length) != 0) {
+                return -1;
+            }
+            continue;
+        case 1:
+            length = 8;
+            break;
+        case 2:
+            if (ctr_protobuf_read_varint(frame, &length) != 0) {
+                return -1;
+            }
+            break;
+        case 5:
+            length = 4;
+            break;
+        default:
+            /* Groups are unsupported by protobuf-c. */
+            return -1;
+        }
+
+        if (length > (uint64_t) (frame->end - frame->cursor)) {
+            return -1;
+        }
+        message = frame->cursor;
+        frame->cursor += (size_t) length;
+        if ((tag & 7) != 2) {
+            continue;
+        }
+        field = protobuf_c_message_descriptor_get_field(frame->descriptor, tag >> 3);
+        if (field == NULL || field->type != PROTOBUF_C_TYPE_MESSAGE) {
+            continue;
+        }
+        if (depth >= CTR_PROTOBUF_MAX_DEPTH) {
+            return -1;
+        }
+        frames[depth].descriptor = field->descriptor;
+        frames[depth].cursor = message;
+        frames[depth].end = message + (size_t) length;
+        depth++;
+    }
+    return 0;
+}
+
+#endif

tests/CMakeLists.txt

@@ -1,4 +1,6 @@
 set(UNIT_TESTS_FILES
+  protobuf.c
+  variant_depth.c
   decoding.c
   basic.c
   span.c

tests/opentelemetry.c

@@ -694,7 +694,71 @@ void test_otlp_decode_invalid_kind()
     free(wire);
 }
 
+
+static void check_otlp_depth(size_t depth, int shape)
+{
+    struct ctrace *original;
+    struct ctrace *decoded;
+    struct ctrace_resource_span *resource;
+    struct cfl_variant *value;
+    struct cfl_variant *parent;
+    struct cfl_kvlist *map;
+    struct cfl_array *array;
+    cfl_sds_t wire;
+    size_t index;
+    size_t offset;
+    int result;
+
+    original = build_sample_trace();
+    TEST_ASSERT(original != NULL);
+    resource = cfl_list_entry(original->resource_spans.next, struct ctrace_resource_span, _head);
+    value = cfl_variant_create_from_string("leaf");
+    TEST_ASSERT(value != NULL);
+    for (index = 0; index < depth; index++) {
+        if (shape == 0 || (shape == 2 && index % 2 == 0)) {
+            map = cfl_kvlist_create();
+            TEST_ASSERT(map != NULL);
+            TEST_ASSERT(cfl_kvlist_insert(map, "k", value) == 0);
+            parent = cfl_variant_create_from_kvlist(map);
+        }
+        else {
+            array = cfl_array_create(1);
+            TEST_ASSERT(array != NULL);
+            TEST_ASSERT(cfl_array_append(array, value) == 0);
+            parent = cfl_variant_create_from_array(array);
+        }
+        TEST_ASSERT(parent != NULL);
+        value = parent;
+    }
+    TEST_ASSERT(cfl_kvlist_insert(resource->resource->attr->kv, "deep", value) == 0);
+    wire = ctr_encode_opentelemetry_create(original);
+    TEST_ASSERT(wire != NULL);
+    ctr_destroy(original);
+    decoded = NULL;
+    offset = 0;
+    result = ctr_decode_opentelemetry_create(&decoded,  wire,
+                                               cfl_sds_len(wire), &offset);
+    TEST_CHECK((result == 0) == (depth < 32));
+    if (decoded != NULL) {
+        ctr_destroy(decoded);
+    }
+    ctr_encode_opentelemetry_destroy(wire);
+}
+
+static void test_otlp_depth_boundary(void)
+{
+    int shape;
+
+    for (shape = 0; shape < 3; shape++) {
+        check_otlp_depth(8, shape);
+        check_otlp_depth(31, shape);
+        check_otlp_depth(32, shape);
+        check_otlp_depth(400, shape);
+    }
+}
+
 TEST_LIST = {
+    {"otlp_depth_boundary", test_otlp_depth_boundary},
     {"otlp_roundtrip",                  test_otlp_roundtrip},
     {"otlp_minimal_trace",              test_otlp_minimal_trace},
     {"otlp_empty_context",              test_otlp_empty_context},

tests/protobuf.c

@@ -0,0 +1,62 @@
+#include "ctr_tests.h"
+#include "../src/ctr_protobuf.h"
+
+static void test_protobuf_depth(void)
+{
+    Opentelemetry__Proto__Common__V1__AnyValue values[51];
+    Opentelemetry__Proto__Common__V1__ArrayValue arrays[50];
+    Opentelemetry__Proto__Common__V1__AnyValue *children[50];
+    unsigned char buffer[1024];
+    size_t size;
+    int index;
+
+    for (index = 0; index < 51; index++) {
+        opentelemetry__proto__common__v1__any_value__init(&values[index]);
+    }
+    values[50].value_case = OPENTELEMETRY__PROTO__COMMON__V1__ANY_VALUE__VALUE_INT_VALUE;
+    values[50].int_value = 7;
+    for (index = 49; index >= 0; index--) {
+        opentelemetry__proto__common__v1__array_value__init(&arrays[index]);
+        children[index] = &values[index + 1];
+        arrays[index].values = &children[index];
+        arrays[index].n_values = 1;
+        values[index].value_case = OPENTELEMETRY__PROTO__COMMON__V1__ANY_VALUE__VALUE_ARRAY_VALUE;
+        values[index].array_value = &arrays[index];
+    }
+
+    /* 99, 100 and 101 schema messages, including scalar AnyValue leaves. */
+    size = opentelemetry__proto__common__v1__any_value__pack(&values[1], buffer);
+    TEST_CHECK(ctr_protobuf_validate(values[1].base.descriptor, buffer, size) == 0);
+    size = opentelemetry__proto__common__v1__array_value__pack(&arrays[0], buffer);
+    TEST_CHECK(ctr_protobuf_validate(arrays[0].base.descriptor, buffer, size) == 0);
+    size = opentelemetry__proto__common__v1__any_value__pack(&values[0], buffer);
+    TEST_CHECK(ctr_protobuf_validate(values[0].base.descriptor, buffer, size) != 0);
+}
+
+static void test_protobuf_wire_boundaries(void)
+{
+    const ProtobufCMessageDescriptor *descriptor;
+    unsigned char truncated[] = {0x2a, 0x02, 0x0a};
+    unsigned char overflow[] = {0x2a, 0xff, 0xff, 0xff, 0xff, 0xff,
+                               0xff, 0xff, 0xff, 0xff, 0x02};
+    unsigned char opaque[] = {0x0a, 0x03, 0xff, 0xff, 0xff};
+    unsigned char unknown[] = {0x7a, 0x03, 0xff, 0xff, 0xff};
+    unsigned char zero_tag[] = {0x00, 0x00};
+    unsigned char fixed[] = {0x21, 0, 0, 0, 0, 0, 0, 0, 0};
+
+    descriptor = &opentelemetry__proto__common__v1__any_value__descriptor;
+    TEST_CHECK(ctr_protobuf_validate(descriptor, NULL, 0) == 0);
+    TEST_CHECK(ctr_protobuf_validate(descriptor, truncated, sizeof(truncated)) != 0);
+    TEST_CHECK(ctr_protobuf_validate(descriptor, overflow, sizeof(overflow)) != 0);
+    TEST_CHECK(ctr_protobuf_validate(descriptor, opaque, sizeof(opaque)) == 0);
+    TEST_CHECK(ctr_protobuf_validate(descriptor, unknown, sizeof(unknown)) == 0);
+    TEST_CHECK(ctr_protobuf_validate(descriptor, zero_tag, sizeof(zero_tag)) != 0);
+    TEST_CHECK(ctr_protobuf_validate(descriptor, fixed, sizeof(fixed)) == 0);
+    TEST_CHECK(ctr_protobuf_validate(descriptor, fixed, sizeof(fixed) - 1) != 0);
+}
+
+TEST_LIST = {
+    {"protobuf_depth", test_protobuf_depth},
+    {"protobuf_wire_boundaries", test_protobuf_wire_boundaries},
+    {0}
+};