Add HTTP body capture for WSGI

.changelog/5272.added

@@ -0,0 +1 @@
+`opentelemetry-util-http`, `opentelemetry-instrumentation-wsgi`: add opt-in request body capture as `http.request.body.content`

instrumentation/opentelemetry-instrumentation-wsgi/src/opentelemetry/instrumentation/wsgi/__init__.py

@@ -207,6 +207,27 @@ def response_hook(
 Note:
     The environment variable names used to capture HTTP headers are still experimental, and thus are subject to change.
 
+Capture HTTP request body
+*************************
+To capture the HTTP request body in the ``http.request.body.content`` span attribute, pass
+``captured_request_body_size``, the maximum number of bytes of the body to capture. Default is ``0``,
+meaning capture is disabled.
+
+.. code-block:: python
+
+    from opentelemetry.instrumentation.wsgi import OpenTelemetryMiddleware
+
+    OpenTelemetryMiddleware(app, captured_request_body_size=4096)
+
+Only the bytes the application reads from ``wsgi.input`` are captured, and only on SERVER spans. The body is
+recorded as a string for textual content types, else as bytes. Bodies with a ``Content-Encoding`` are not captured.
+
+The body is not captured when ``OTEL_SEMCONV_STABILITY_OPT_IN`` is set to ``http``, since the attribute is not stable.
+
+Note:
+    The request body may contain sensitive data. This feature and the attribute are experimental, and thus are
+    subject to change.
+
 Sanitizing methods
 ******************
 In order to prevent unbound cardinality for HTTP methods by default nonstandard ones are labeled as ``NONSTANDARD``.
@@ -284,6 +305,12 @@ def response_hook(
     redact_url,
     sanitize_method,
 )
+from opentelemetry.util.http.body_capture import (
+    HTTP_REQUEST_BODY_CONTENT,
+    BodyBuffer,
+    CapturingStream,
+    create_body_buffer,
+)
 
 if TYPE_CHECKING:
     from wsgiref.types import StartResponse, WSGIApplication, WSGIEnvironment
@@ -573,6 +600,9 @@ class OpenTelemetryMiddleware:
                          globally configured one is used.
         meter_provider: Optional meter provider to use. If omitted the current
                          globally configured one is used.
+        captured_request_body_size: Maximum number of bytes of the request body
+                                    to capture as a span attribute.
+                                    ``0`` disables capture.
     """
 
     def __init__(
@@ -582,6 +612,7 @@ def __init__(
         response_hook: ResponseHook | None = None,
         tracer_provider: TracerProvider | None = None,
         meter_provider: MeterProvider | None = None,
+        captured_request_body_size: int = 0,
     ):
         # initialize semantic conventions opt-in if needed
         _OpenTelemetrySemanticConventionStability._initialize()
@@ -626,6 +657,24 @@ def __init__(
         self.request_hook = request_hook
         self.response_hook = response_hook
         self._sem_conv_opt_in_mode = sem_conv_opt_in_mode
+        self._captured_request_body_size = (
+            0 if sem_conv_opt_in_mode == _StabilityMode.HTTP else captured_request_body_size
+        )
+
+    def _wrap_request_body(self, span: trace.Span, environ: WSGIEnvironment) -> BodyBuffer | None:
+        if not (self._captured_request_body_size > 0 and span.is_recording() and span.kind == trace.SpanKind.SERVER):
+            return None
+        stream = environ.get("wsgi.input")
+        if stream is None:
+            return None
+        body_buffer = create_body_buffer(
+            self._captured_request_body_size,
+            environ.get("CONTENT_TYPE"),
+            environ.get("HTTP_CONTENT_ENCODING"),
+        )
+        if body_buffer is not None:
+            environ["wsgi.input"] = CapturingStream(stream, body_buffer)
+        return body_buffer
 
     @staticmethod
     def _create_start_response(
@@ -688,6 +737,7 @@ def __call__(self, environ: WSGIEnvironment, start_response: StartResponse):
             context_getter=wsgi_getter,
             attributes=req_attrs,
         )
+        body_buffer = self._wrap_request_body(span, environ)
 
         if self.request_hook:
             self.request_hook(span, environ)
@@ -708,13 +758,14 @@ def __call__(self, environ: WSGIEnvironment, start_response: StartResponse):
                     self._sem_conv_opt_in_mode,
                 )
                 iterable = self.wsgi(environ, start_response)
-                return _end_span_after_iterating(iterable, span, token)
+                return _end_span_after_iterating(iterable, span, token, body_buffer)
         except Exception as ex:
             if _report_new(self._sem_conv_opt_in_mode):
                 req_attrs[ERROR_TYPE] = type(ex).__qualname__
                 if span.is_recording():
                     span.set_attribute(ERROR_TYPE, type(ex).__qualname__)
                 span.set_status(Status(StatusCode.ERROR, str(ex)))
+            _set_request_body_attribute(span, body_buffer)
             span.end()
             if token is not None:
                 context.detach(token)
@@ -739,17 +790,31 @@ def __call__(self, environ: WSGIEnvironment, start_response: StartResponse):
             self.active_requests_counter.add(-1, active_requests_count_attrs)
 
 
+def _set_request_body_attribute(span: trace.Span, body_buffer: BodyBuffer | None) -> None:
+    if body_buffer is None:
+        return
+    body = body_buffer.to_attribute_value()
+    if body is not None:
+        span.set_attribute(HTTP_REQUEST_BODY_CONTENT, body)
+
+
 # Put this in a subfunction to not delay the call to the wrapped
 # WSGI application (instrumentation should change the application
 # behavior as little as possible).
-def _end_span_after_iterating(iterable: Iterable[T], span: trace.Span, token: object) -> Iterable[T]:
+def _end_span_after_iterating(
+    iterable: Iterable[T],
+    span: trace.Span,
+    token: object,
+    body_buffer: BodyBuffer | None,
+) -> Iterable[T]:
     try:
         with trace.use_span(span):
             yield from iterable
     finally:
         close = getattr(iterable, "close", None)
         if close:
             close()
+        _set_request_body_attribute(span, body_buffer)
         span.end()
         if token is not None:
             context.detach(token)

instrumentation/opentelemetry-instrumentation-wsgi/tests/test_wsgi_middleware.py

@@ -5,6 +5,7 @@
 
 from __future__ import annotations
 
+import io
 import sys
 import unittest
 import wsgiref.util as wsgiref_util
@@ -78,6 +79,7 @@
     OTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SERVER_RESPONSE,
     OTEL_PYTHON_INSTRUMENTATION_HTTP_CAPTURE_ALL_METHODS,
 )
+from opentelemetry.util.http.body_capture import HTTP_REQUEST_BODY_CONTENT
 
 
 class Response:
@@ -1333,5 +1335,144 @@ def test_repeat_custom_response_headers_added_in_server_span(self):
         self.assertSpanHasAttributes(span, expected)
 
 
+def read_body_wsgi(environ, start_response):
+    environ["wsgi.input"].read()
+    start_response("200 OK", [("Content-Type", "text/plain")])
+    return [b"*"]
+
+
+def read_body_lazily_wsgi(environ, start_response):
+    start_response("200 OK", [("Content-Type", "text/plain")])
+    environ["wsgi.input"].read()
+    yield b"*"
+
+
+def read_body_then_raise_wsgi(environ, start_response):
+    environ["wsgi.input"].read()
+    raise ValueError
+
+
+class TestWsgiRequestBodyCapture(WsgiTestBase):
+    def setUp(self):
+        super().setUp()
+        self.set_sem_conv_mode("default")
+        self.environ["REQUEST_METHOD"] = "POST"
+        self.set_body(b'{"hello": "world"}', "application/json")
+
+    def tearDown(self):
+        super().tearDown()
+        _OpenTelemetrySemanticConventionStability._initialized = False
+
+    def set_sem_conv_mode(self, mode):
+        env_patch = mock.patch.dict("os.environ", {OTEL_SEMCONV_STABILITY_OPT_IN: mode})
+        env_patch.start()
+        self.addCleanup(env_patch.stop)
+        _OpenTelemetrySemanticConventionStability._initialized = False
+
+    def set_body(self, body, content_type):
+        self.environ["wsgi.input"] = io.BytesIO(body)
+        self.environ["CONTENT_LENGTH"] = str(len(body))
+        self.environ["CONTENT_TYPE"] = content_type
+
+    def get_span(self, app, **kwargs):
+        middleware = otel_wsgi.OpenTelemetryMiddleware(app, **kwargs)
+        response = middleware(self.environ, self.start_response)
+        self.assertEqual(list(response), [b"*"])
+        span_list = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(span_list), 1)
+        return span_list[0]
+
+    def test_not_captured_by_default(self):
+        original_input = self.environ["wsgi.input"]
+        span = self.get_span(read_body_wsgi)
+        self.assertNotIn(HTTP_REQUEST_BODY_CONTENT, span.attributes)
+        self.assertIs(self.environ["wsgi.input"], original_input)
+
+    def test_not_captured_with_non_positive_size(self):
+        original_input = self.environ["wsgi.input"]
+        for size in (0, -1):
+            with self.subTest(size=size):
+                self.memory_exporter.clear()
+                span = self.get_span(read_body_wsgi, captured_request_body_size=size)
+                self.assertNotIn(HTTP_REQUEST_BODY_CONTENT, span.attributes)
+                self.assertIs(self.environ["wsgi.input"], original_input)
+
+    def test_text_body_captured_as_string(self):
+        span = self.get_span(read_body_wsgi, captured_request_body_size=1024)
+        self.assertEqual(span.attributes[HTTP_REQUEST_BODY_CONTENT], '{"hello": "world"}')
+        self.assertIsInstance(span.attributes[HTTP_REQUEST_BODY_CONTENT], str)
+
+    def test_binary_body_captured_as_bytes(self):
+        self.set_body(b"\x00\x01\xff", "application/octet-stream")
+        span = self.get_span(read_body_wsgi, captured_request_body_size=1024)
+        self.assertEqual(span.attributes[HTTP_REQUEST_BODY_CONTENT], b"\x00\x01\xff")
+        self.assertIsInstance(span.attributes[HTTP_REQUEST_BODY_CONTENT], bytes)
+
+    def test_body_truncated_to_captured_request_body_size(self):
+        span = self.get_span(read_body_wsgi, captured_request_body_size=6)
+        self.assertEqual(span.attributes[HTTP_REQUEST_BODY_CONTENT], '{"hell')
+
+    def test_application_reads_full_body_past_limit(self):
+        def app(environ, start_response):
+            self.assertEqual(environ["wsgi.input"].read(), b'{"hello": "world"}')
+            return read_body_wsgi(environ, start_response)
+
+        span = self.get_span(app, captured_request_body_size=6)
+        self.assertEqual(span.attributes[HTTP_REQUEST_BODY_CONTENT], '{"hell')
+
+    def test_unread_body_not_captured(self):
+        span = self.get_span(simple_wsgi, captured_request_body_size=1024)
+        self.assertNotIn(HTTP_REQUEST_BODY_CONTENT, span.attributes)
+        self.assertEqual(self.environ["wsgi.input"].tell(), 0)
+
+    def test_body_read_while_iterating_response(self):
+        span = self.get_span(read_body_lazily_wsgi, captured_request_body_size=1024)
+        self.assertEqual(span.attributes[HTTP_REQUEST_BODY_CONTENT], '{"hello": "world"}')
+
+    def test_body_captured_when_application_raises(self):
+        app = otel_wsgi.OpenTelemetryMiddleware(read_body_then_raise_wsgi, captured_request_body_size=1024)
+        with self.assertRaises(ValueError):
+            app(self.environ, self.start_response)
+        span = self.memory_exporter.get_finished_spans()[0]
+        self.assertEqual(span.attributes[HTTP_REQUEST_BODY_CONTENT], '{"hello": "world"}')
+
+    def test_content_encoded_body_not_captured(self):
+        self.environ["HTTP_CONTENT_ENCODING"] = "gzip"
+        original_input = self.environ["wsgi.input"]
+        span = self.get_span(read_body_wsgi, captured_request_body_size=1024)
+        self.assertNotIn(HTTP_REQUEST_BODY_CONTENT, span.attributes)
+        self.assertIs(self.environ["wsgi.input"], original_input)
+
+    def test_not_captured_on_internal_span(self):
+        tracer = self.tracer_provider.get_tracer(__name__)
+        original_input = self.environ["wsgi.input"]
+        with tracer.start_as_current_span("test", kind=trace_api.SpanKind.SERVER):
+            app = otel_wsgi.OpenTelemetryMiddleware(read_body_wsgi, captured_request_body_size=1024)
+            self.assertEqual(list(app(self.environ, self.start_response)), [b"*"])
+        span = self.memory_exporter.get_finished_spans()[0]
+        self.assertEqual(span.kind, trace_api.SpanKind.INTERNAL)
+        self.assertNotIn(HTTP_REQUEST_BODY_CONTENT, span.attributes)
+        self.assertIs(self.environ["wsgi.input"], original_input)
+
+    def test_not_captured_when_not_recording(self):
+        original_input = self.environ["wsgi.input"]
+        app = otel_wsgi.OpenTelemetryMiddleware(
+            read_body_wsgi,
+            tracer_provider=trace_api.NoOpTracerProvider(),
+            captured_request_body_size=1024,
+        )
+        self.assertEqual(list(app(self.environ, self.start_response)), [b"*"])
+        self.assertIs(self.environ["wsgi.input"], original_input)
+
+    def test_sem_conv_modes(self):
+        for mode, captured in (("http", False), ("http/dup", True)):
+            with self.subTest(mode=mode):
+                self.memory_exporter.clear()
+                self.set_sem_conv_mode(mode)
+                self.set_body(b"hello", "text/plain")
+                span = self.get_span(read_body_wsgi, captured_request_body_size=1024)
+                self.assertEqual(HTTP_REQUEST_BODY_CONTENT in span.attributes, captured)
+
+
 if __name__ == "__main__":
     unittest.main()

util/opentelemetry-util-http/pyproject.toml

@@ -24,6 +24,9 @@ classifiers = [
   "Programming Language :: Python :: 3.13",
   "Programming Language :: Python :: 3.14",
 ]
+dependencies = [
+  "wrapt >= 1.0.0, < 3.0.0",
+]
 
 [project.urls]
 Homepage = "https://github.com/open-telemetry/opentelemetry-python-contrib/tree/main/util/opentelemetry-util-http"