opentelemetry-propagator-jaeger: enforce baggage limits on extract

.changelog/5556.fixed

@@ -0,0 +1 @@
+`opentelemetry-propagator-jaeger`: enforce baggage limits on both `uberctx-` extract and inject, borrowing the same limits (180 entries, 4096 bytes per entry, 8192 bytes total) the package's core `W3CBaggagePropagator` already uses, so neither an inbound carrier nor an in-process baggage map can produce unbounded work or headers.

propagator/opentelemetry-propagator-jaeger/src/opentelemetry/propagators/jaeger/__init__.py

@@ -2,6 +2,7 @@
 # SPDX-License-Identifier: Apache-2.0
 
 import collections.abc
+import itertools
 import urllib.parse
 
 from opentelemetry import baggage, trace
@@ -26,6 +27,11 @@ class JaegerPropagator(TextMapPropagator):
     TRACE_ID_KEY = "uber-trace-id"
     BAGGAGE_PREFIX = "uberctx-"
     DEBUG_FLAG = 0x02
+    # The Jaeger format defines no baggage limits, so the W3C Baggage spec
+    # limits are borrowed to bound an unbounded inbound carrier on extract.
+    MAX_BAGGAGE_ENTRIES = 180
+    MAX_BAGGAGE_ENTRY_BYTES = 4096
+    MAX_BAGGAGE_TOTAL_BYTES = 8192
 
     def extract(
         self,
@@ -90,9 +96,7 @@ def inject(
         baggage_entries = baggage.get_all(context=context)
         if not baggage_entries:
             return
-        for key, value in baggage_entries.items():
-            baggage_key = self.BAGGAGE_PREFIX + key
-            setter.set(carrier, baggage_key, urllib.parse.quote(str(value)))
+        self._inject_baggage(setter, carrier, baggage_entries)
 
     @property
     def fields(self) -> set[str]:
@@ -104,18 +108,61 @@ def _extract_baggage(
         carrier: CarrierT,
         context: Context,
     ) -> Context:
-        baggage_keys = [key for key in getter.keys(carrier) if key.startswith(self.BAGGAGE_PREFIX)]
-        for key in baggage_keys:
+        # The limit bounds the candidates inspected, not the entries kept, so a
+        # carrier full of oversized ones cannot force unbounded work.
+        candidates = itertools.islice(
+            (key for key in getter.keys(carrier) if key.startswith(self.BAGGAGE_PREFIX)),
+            self.MAX_BAGGAGE_ENTRIES,
+        )
+        pairs = []
+        for key in candidates:
             value = _extract_first_element(getter.get(carrier, key))
-            if value is None:
-                continue
+            if value is not None:
+                pairs.append((key.replace(self.BAGGAGE_PREFIX, ""), value))
+
+        for baggage_key, value in _limit_baggage_bytes(
+            pairs, self.MAX_BAGGAGE_ENTRY_BYTES, self.MAX_BAGGAGE_TOTAL_BYTES
+        ):
             context = baggage.set_baggage(
-                key.replace(self.BAGGAGE_PREFIX, ""),
+                baggage_key,
                 urllib.parse.unquote(value).strip(),
                 context=context,
             )
         return context
 
+    def _inject_baggage(
+        self,
+        setter: Setter[CarrierT],
+        carrier: CarrierT,
+        baggage_entries: collections.abc.Mapping[str, object],
+    ) -> None:
+        candidates = itertools.islice(baggage_entries.items(), self.MAX_BAGGAGE_ENTRIES)
+        pairs = [(key, urllib.parse.quote(str(value))) for key, value in candidates]
+
+        for key, encoded_value in _limit_baggage_bytes(
+            pairs, self.MAX_BAGGAGE_ENTRY_BYTES, self.MAX_BAGGAGE_TOTAL_BYTES
+        ):
+            setter.set(carrier, self.BAGGAGE_PREFIX + key, encoded_value)
+
+
+def _limit_baggage_bytes(
+    pairs: collections.abc.Iterable[tuple[str, str]],
+    max_entry_bytes: int,
+    max_total_bytes: int,
+) -> collections.abc.Iterator[tuple[str, str]]:
+    total_bytes = 0
+    accepted = 0
+    for key, value in pairs:
+        entry_bytes = len(key.encode()) + len(value.encode()) + 1
+        if entry_bytes > max_entry_bytes:
+            continue
+        separator_bytes = 1 if accepted > 0 else 0
+        if total_bytes + separator_bytes + entry_bytes > max_total_bytes:
+            continue
+        yield key, value
+        total_bytes += separator_bytes + entry_bytes
+        accepted += 1
+
 
 def _format_uber_trace_id(trace_id, span_id, parent_span_id, flags):
     return f"{format_trace_id(trace_id)}:{format_span_id(span_id)}:{format_span_id(parent_span_id)}:{flags:02x}"

propagator/opentelemetry-propagator-jaeger/tests/test_jaeger_propagator.py

@@ -10,6 +10,7 @@
 from opentelemetry.propagators import (  # pylint: disable=no-name-in-module
     jaeger,
 )
+from opentelemetry.propagators.textmap import DefaultGetter
 from opentelemetry.sdk import trace
 from opentelemetry.sdk.trace import id_generator
 from opentelemetry.test import TestCase
@@ -46,6 +47,8 @@ def get_context_new_carrier(old_carrier, carrier_baggage=None):
 
 
 class TestJaegerPropagator(TestCase):
+    # pylint: disable=too-many-public-methods
+
     @classmethod
     def setUpClass(cls):
         generator = id_generator.RandomIdGenerator()
@@ -134,6 +137,116 @@ def test_extract_empty_baggage_value(self):
         context = FORMAT.extract(old_carrier)
         self.assertDictEqual({"key3": "value3"}, context[_BAGGAGE_KEY])
 
+    def test_extract_enforces_max_baggage_entries(self):
+        old_carrier = {FORMAT.TRACE_ID_KEY: self.serialized_uber_trace_id}
+        for index in range(200):
+            old_carrier[f"uberctx-k{index}"] = f"v{index}"
+        extracted = FORMAT.extract(old_carrier)[_BAGGAGE_KEY]
+        self.assertEqual(FORMAT.MAX_BAGGAGE_ENTRIES, len(extracted))
+        self.assertIn("k0", extracted)
+        self.assertNotIn("k180", extracted)
+
+    def test_extract_drops_oversized_baggage_entry(self):
+        old_carrier = {
+            FORMAT.TRACE_ID_KEY: self.serialized_uber_trace_id,
+            "uberctx-ok": "value",
+            "uberctx-big": "x" * 5000,
+        }
+        context = FORMAT.extract(old_carrier)
+        self.assertDictEqual({"ok": "value"}, context[_BAGGAGE_KEY])
+
+    def test_extract_measures_entry_limit_in_bytes(self):
+        # 2100 multibyte characters is 4200 bytes: under the character limit,
+        # over the byte limit, so the entry must be dropped.
+        old_carrier = {
+            FORMAT.TRACE_ID_KEY: self.serialized_uber_trace_id,
+            "uberctx-ok": "value",
+            "uberctx-u": "é" * 2100,
+        }
+        context = FORMAT.extract(old_carrier)
+        self.assertDictEqual({"ok": "value"}, context[_BAGGAGE_KEY])
+
+    def test_extract_enforces_max_baggage_total_bytes(self):
+        old_carrier = {FORMAT.TRACE_ID_KEY: self.serialized_uber_trace_id}
+        for index in range(100):
+            old_carrier[f"uberctx-k{index}"] = "y" * 200
+        extracted = FORMAT.extract(old_carrier)[_BAGGAGE_KEY]
+        self.assertLess(len(extracted), 100)
+        self.assertIn("k0", extracted)
+        self.assertNotIn("k99", extracted)
+
+    def test_extract_counts_the_key_value_separator_byte(self):
+        old_carrier = {
+            FORMAT.TRACE_ID_KEY: self.serialized_uber_trace_id,
+            "uberctx-fits": "x" * (FORMAT.MAX_BAGGAGE_ENTRY_BYTES - len("fits") - 1),
+            "uberctx-over": "x" * (FORMAT.MAX_BAGGAGE_ENTRY_BYTES - len("over")),
+        }
+        extracted = FORMAT.extract(old_carrier)[_BAGGAGE_KEY]
+        self.assertIn("fits", extracted)
+        self.assertNotIn("over", extracted)
+
+    def test_extract_charges_a_separator_byte_between_accepted_entries(self):
+        small_entry_bytes = len("small") + 1  # empty value
+        filler_a_entry_bytes = FORMAT.MAX_BAGGAGE_ENTRY_BYTES
+        filler_a_value_len = filler_a_entry_bytes - len("filler_a") - 1
+        target_total_after_fillers = FORMAT.MAX_BAGGAGE_TOTAL_BYTES - small_entry_bytes
+        filler_b_entry_bytes = target_total_after_fillers - filler_a_entry_bytes - 1
+        filler_b_value_len = filler_b_entry_bytes - len("filler_b") - 1
+        old_carrier = {
+            FORMAT.TRACE_ID_KEY: self.serialized_uber_trace_id,
+            "uberctx-filler_a": "x" * filler_a_value_len,
+            "uberctx-filler_b": "x" * filler_b_value_len,
+            "uberctx-small": "",
+        }
+        extracted = FORMAT.extract(old_carrier)[_BAGGAGE_KEY]
+        self.assertIn("filler_a", extracted)
+        self.assertIn("filler_b", extracted)
+        self.assertNotIn("small", extracted)
+
+    def test_extract_stops_inspecting_after_the_candidate_limit(self):
+        old_carrier = {FORMAT.TRACE_ID_KEY: self.serialized_uber_trace_id}
+        for index in range(1000):
+            old_carrier[f"uberctx-k{index}"] = "x" * 5000
+
+        class CountingGetter(DefaultGetter):
+            def __init__(self):
+                self.reads = 0
+
+            def get(self, carrier, key):
+                self.reads += 1
+                return super().get(carrier, key)
+
+        getter = CountingGetter()
+        FORMAT.extract(old_carrier, getter=getter)
+        self.assertLessEqual(getter.reads, FORMAT.MAX_BAGGAGE_ENTRIES + 1)
+
+    def test_inject_enforces_max_baggage_entries(self):
+        span = trace_api.NonRecordingSpan(trace_api.SpanContext(1, 1, True))
+        ctx = trace_api.set_span_in_context(span)
+        for index in range(200):
+            ctx = baggage.set_baggage(f"k{index}", f"v{index}", ctx)
+
+        carrier = {}
+        FORMAT.inject(carrier, context=ctx)
+
+        self.assertEqual(FORMAT.MAX_BAGGAGE_ENTRIES, sum(1 for key in carrier if key.startswith(FORMAT.BAGGAGE_PREFIX)))
+        self.assertIn(FORMAT.BAGGAGE_PREFIX + "k0", carrier)
+        self.assertNotIn(FORMAT.BAGGAGE_PREFIX + "k180", carrier)
+
+    def test_inject_enforces_max_baggage_total_bytes(self):
+        span = trace_api.NonRecordingSpan(trace_api.SpanContext(1, 1, True))
+        ctx = trace_api.set_span_in_context(span)
+        for index in range(100):
+            ctx = baggage.set_baggage(f"k{index}", "y" * 200, ctx)
+
+        carrier = {}
+        FORMAT.inject(carrier, context=ctx)
+
+        injected = [key for key in carrier if key.startswith(FORMAT.BAGGAGE_PREFIX)]
+        self.assertLess(len(injected), 100)
+        self.assertIn(FORMAT.BAGGAGE_PREFIX + "k0", carrier)
+        self.assertNotIn(FORMAT.BAGGAGE_PREFIX + "k99", carrier)
+
     def test_extract_invalid_uber_trace_id(self):
         old_carrier = {
             "uber-trace-id": "000000000000000000000000deadbeef:00000000deadbef0:00",