diff --git a/.changelog/5272.added b/.changelog/5272.added new file mode 100644 index 00000000000..3ba46c661ea --- /dev/null +++ b/.changelog/5272.added @@ -0,0 +1 @@ +`opentelemetry-util-http`, `opentelemetry-instrumentation-wsgi`: add opt-in request body capture as `http.request.body.content` diff --git a/instrumentation/opentelemetry-instrumentation-wsgi/src/opentelemetry/instrumentation/wsgi/__init__.py b/instrumentation/opentelemetry-instrumentation-wsgi/src/opentelemetry/instrumentation/wsgi/__init__.py index d7671362ac5..52f54ce6a5c 100644 --- a/instrumentation/opentelemetry-instrumentation-wsgi/src/opentelemetry/instrumentation/wsgi/__init__.py +++ b/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,10 +790,23 @@ 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 @@ -750,6 +814,7 @@ def _end_span_after_iterating(iterable: Iterable[T], span: trace.Span, token: ob 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) diff --git a/instrumentation/opentelemetry-instrumentation-wsgi/tests/test_wsgi_middleware.py b/instrumentation/opentelemetry-instrumentation-wsgi/tests/test_wsgi_middleware.py index a49a35d3deb..ecbd5f1cb81 100644 --- a/instrumentation/opentelemetry-instrumentation-wsgi/tests/test_wsgi_middleware.py +++ b/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() diff --git a/util/opentelemetry-util-http/pyproject.toml b/util/opentelemetry-util-http/pyproject.toml index 48170a2f319..d24a08a821c 100644 --- a/util/opentelemetry-util-http/pyproject.toml +++ b/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" diff --git a/util/opentelemetry-util-http/src/opentelemetry/util/http/body_capture.py b/util/opentelemetry-util-http/src/opentelemetry/util/http/body_capture.py new file mode 100644 index 00000000000..91e3652c2bd --- /dev/null +++ b/util/opentelemetry-util-http/src/opentelemetry/util/http/body_capture.py @@ -0,0 +1,190 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +"""Shared helpers for capturing HTTP bodies as span attributes. + +Instrumentation does not itself read the body, so only the bytes read by the +application can be recorded. + +This feature is experimental. +""" + +from __future__ import annotations + +import codecs +import functools +from collections.abc import Callable, Iterator +from typing import Any, BinaryIO, Final + +import wrapt + +# TODO: use the semconv constant once the semconv package is generated from a +# semantic conventions release that includes this attribute (> 1.44.0). +HTTP_REQUEST_BODY_CONTENT: Final = "http.request.body.content" + +_TEXTUAL_SUBTYPE_SUFFIXES = ("/json", "+json", "/xml", "+xml", "/yaml", "+yaml") + + +def _parse_content_type(content_type: str) -> tuple[str, str | None]: + """Returns the lowercased type/subtype and the charset parameter, if any.""" + media_type, *params = content_type.split(";") + charset = None + for param in params: + name, _, value = param.partition("=") + if name.strip().lower() == "charset": + charset = value.strip().strip('"') or None + return media_type.strip().lower(), charset + + +def _is_textual(media_type: str) -> bool: + return ( + media_type.startswith("text/") + or media_type.endswith(_TEXTUAL_SUBTYPE_SUFFIXES) + or media_type == "application/x-www-form-urlencoded" + ) + + +class BodyBuffer: + """Holds up to ``max_size`` bytes of a body, indexed by stream position. + + Writes are keyed by the position the bytes were read from, so a body that + is seeked back and read again is not recorded twice. + """ + + def __init__(self, max_size: int, content_type: str | None): + self._max_size = max_size + self._content_type = content_type + self._data = bytearray() + + def write(self, position: int, data: bytes | bytearray | memoryview) -> None: + if position > len(self._data): + # Bytes before this position were skipped by the application, so + # the captured prefix cannot be extended contiguously. + return + end = min(position + len(data), self._max_size) + self._data[position:end] = data[: end - position] + + def to_attribute_value(self) -> str | bytes | None: + """Returns the captured body as a string when it is text, else bytes. + + Returns ``None`` when nothing was captured. + """ + if not self._data: + return None + data = bytes(self._data) + if self._content_type is None: + return data + media_type, charset = _parse_content_type(self._content_type) + if charset is None: + if not _is_textual(media_type): + return data + charset = "utf-8" + try: + decoder = codecs.getincrementaldecoder(charset)(errors="strict") + # The captured bytes may end in the middle of a character, because + # of the size limit or because the application stopped reading. + # final=False drops that partial character so the value is valid. + return decoder.decode(data, final=False) + except (LookupError, UnicodeDecodeError): + return data + + +def create_body_buffer( + max_size: int, + content_type: str | None, + content_encoding: str | None, +) -> BodyBuffer | None: + """Returns a buffer for a body, or ``None`` if it should not be captured. + + Bodies with a content coding are not captured. + """ + if max_size <= 0: + return None + if content_encoding and content_encoding.strip().lower() != "identity": + return None + return BodyBuffer(max_size, content_type) + + +# Stream methods that do not move the stream position. +_NON_CONSUMING_METHODS = frozenset( + {"close", "fileno", "flush", "isatty", "peek", "readable", "seekable", "tell", "writable"} +) + + +class CapturingStream(wrapt.ObjectProxy): + """Proxies a binary input stream, copying what the application reads. + + Reads through the PEP-defined WSGI input stream methods, iteration, ``readinto`` and ``seek`` + are captured. Calling any other method that is not known to leave the stream position alone + stops capture, since it may consume bytes the proxy cannot see. + """ + + def __init__(self, stream: BinaryIO, buffer: BodyBuffer): + super().__init__(stream) + self._self_buffer = buffer + self._self_position: int | None = 0 + + def _self_capture(self, data: bytes | bytearray | memoryview) -> None: + if self._self_position is None: + return + self._self_buffer.write(self._self_position, data) + self._self_position += len(data) + + def read(self, *args: Any) -> bytes: + data = self.__wrapped__.read(*args) + self._self_capture(data) + return data + + def readline(self, *args: Any) -> bytes: + data = self.__wrapped__.readline(*args) + self._self_capture(data) + return data + + def readlines(self, *args: Any) -> list[bytes]: + lines = self.__wrapped__.readlines(*args) + for line in lines: + self._self_capture(line) + return lines + + def __iter__(self) -> Iterator[bytes]: + for data in self.__wrapped__: + self._self_capture(data) + yield data + + def __next__(self) -> bytes: + data = next(self.__wrapped__) + self._self_capture(data) + return data + + def _self_readinto(self, target: bytearray | memoryview) -> int | None: + count = self.__wrapped__.readinto(target) + if count: + self._self_capture(memoryview(target)[:count]) + return count + + def _self_seek(self, *args: Any) -> int: + position = self.__wrapped__.seek(*args) + # Without a known position, further reads cannot be placed in the + # buffer, so stop capturing instead of corrupting it. + self._self_position = position if isinstance(position, int) else None + return position + + def _self_stop_capture_on_call(self, method: Callable[..., Any]) -> Callable[..., Any]: + @functools.wraps(method) + def call(*args: Any, **kwargs: Any) -> Any: + self._self_position = None + return method(*args, **kwargs) + + return call + + def __getattr__(self, name: str) -> Any: + attribute = getattr(self.__wrapped__, name) + # Non-PEP, but commonly used WSGI methods. We don't define them on the class + # itself so they can be queried with hasattr. + if name == "readinto": + return self._self_readinto + if name == "seek": + return self._self_seek + if name in _NON_CONSUMING_METHODS or not callable(attribute): + return attribute + return self._self_stop_capture_on_call(attribute) diff --git a/util/opentelemetry-util-http/tests/test_body_capture.py b/util/opentelemetry-util-http/tests/test_body_capture.py new file mode 100644 index 00000000000..fcb6a667a2a --- /dev/null +++ b/util/opentelemetry-util-http/tests/test_body_capture.py @@ -0,0 +1,260 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +import io +import unittest + +from opentelemetry.util.http.body_capture import ( + BodyBuffer, + CapturingStream, + create_body_buffer, +) + + +class TestBodyBuffer(unittest.TestCase): + def test_empty_returns_none(self): + self.assertIsNone(BodyBuffer(10, "text/plain").to_attribute_value()) + + def test_truncates_to_max_size(self): + buffer = BodyBuffer(5, "text/plain") + buffer.write(0, b"hello world") + self.assertEqual(buffer.to_attribute_value(), "hello") + + def test_rewrite_of_same_position_is_not_duplicated(self): + buffer = BodyBuffer(100, "text/plain") + buffer.write(0, b"hello") + buffer.write(0, b"hello") + buffer.write(5, b" world") + self.assertEqual(buffer.to_attribute_value(), "hello world") + + def test_write_after_gap_is_ignored(self): + buffer = BodyBuffer(100, "text/plain") + buffer.write(0, b"ab") + buffer.write(5, b"cd") + self.assertEqual(buffer.to_attribute_value(), "ab") + + def test_textual_content_types_are_strings(self): + for content_type in ( + "text/plain", + "TEXT/HTML", + "application/json", + "application/vnd.api+json", + "application/xml", + "image/svg+xml", + "application/yaml", + "application/vnd.foo+yaml", + "application/x-www-form-urlencoded", + "application/octet-stream; charset=utf-8", + ): + with self.subTest(content_type=content_type): + buffer = BodyBuffer(100, content_type) + buffer.write(0, b"a=1") + self.assertEqual(buffer.to_attribute_value(), "a=1") + + def test_non_textual_content_types_are_bytes(self): + for content_type in ( + "application/octet-stream", + "image/png", + "multipart/form-data; boundary=x", + "application/jsonl", + ): + with self.subTest(content_type=content_type): + buffer = BodyBuffer(100, content_type) + buffer.write(0, b"a=1") + value = buffer.to_attribute_value() + self.assertIsInstance(value, bytes) + self.assertEqual(value, b"a=1") + + def test_missing_content_type_is_bytes(self): + # An empty header value is treated like a missing one. + for content_type in (None, ""): + with self.subTest(content_type=content_type): + buffer = BodyBuffer(100, content_type) + buffer.write(0, b"hello") + value = buffer.to_attribute_value() + self.assertIsInstance(value, bytes) + self.assertEqual(value, b"hello") + + def test_charset_parameter_is_used(self): + for content_type in ( + "text/plain; charset=latin-1", + 'text/plain; CHARSET="latin-1"', + "text/plain;charset=latin-1;format=flowed", + ): + with self.subTest(content_type=content_type): + buffer = BodyBuffer(100, content_type) + buffer.write(0, "héllo".encode("latin-1")) + self.assertEqual(buffer.to_attribute_value(), "héllo") + + def test_unknown_charset_falls_back_to_bytes(self): + buffer = BodyBuffer(100, "text/plain; charset=not-a-charset") + buffer.write(0, b"hello") + self.assertEqual(buffer.to_attribute_value(), b"hello") + + def test_invalid_text_falls_back_to_bytes(self): + buffer = BodyBuffer(100, "application/json") + buffer.write(0, b'{"a": "\xff"}') + self.assertEqual(buffer.to_attribute_value(), b'{"a": "\xff"}') + + def test_truncation_drops_partial_character(self): + # "é" is two bytes in UTF-8; the limit splits it. + buffer = BodyBuffer(2, "text/plain") + buffer.write(0, "aé".encode()) + self.assertEqual(buffer.to_attribute_value(), "a") + + def test_partial_read_drops_partial_character(self): + buffer = BodyBuffer(100, "text/plain") + buffer.write(0, "aé".encode()[:2]) + self.assertEqual(buffer.to_attribute_value(), "a") + + def test_truncation_respects_multibyte_charset(self): + buffer = BodyBuffer(5, "text/plain; charset=utf-16-le") + buffer.write(0, "abc".encode("utf-16-le")) + self.assertEqual(buffer.to_attribute_value(), "ab") + + +class TestCreateBodyBuffer(unittest.TestCase): + def test_creates_buffer(self): + self.assertIsInstance(create_body_buffer(10, "text/plain", None), BodyBuffer) + self.assertIsInstance(create_body_buffer(10, None, ""), BodyBuffer) + self.assertIsInstance(create_body_buffer(10, None, " Identity "), BodyBuffer) + + def test_no_buffer_for_content_coding(self): + self.assertIsNone(create_body_buffer(10, "text/plain", "gzip")) + self.assertIsNone(create_body_buffer(10, "text/plain", "deflate, br")) + + def test_no_buffer_for_non_positive_size(self): + self.assertIsNone(create_body_buffer(0, "text/plain", None)) + self.assertIsNone(create_body_buffer(-1, "text/plain", None)) + + +class TestCapturingStream(unittest.TestCase): + def setUp(self): + self.buffer = BodyBuffer(100, "text/plain") + self.raw = io.BytesIO(b"line one\nline two\nline three\n") + self.stream = CapturingStream(self.raw, self.buffer) + + def test_read_returns_underlying_data(self): + self.assertEqual(self.stream.read(4), b"line") + self.assertEqual(self.stream.read(), b" one\nline two\nline three\n") + self.assertEqual(self.buffer.to_attribute_value(), "line one\nline two\nline three\n") + + def test_readline(self): + self.assertEqual(self.stream.readline(), b"line one\n") + self.assertEqual(self.stream.readline(4), b"line") + self.assertEqual(self.buffer.to_attribute_value(), "line one\nline") + + def test_readlines(self): + self.assertEqual(self.stream.readlines(), [b"line one\n", b"line two\n", b"line three\n"]) + self.assertEqual(self.buffer.to_attribute_value(), "line one\nline two\nline three\n") + + def test_iteration(self): + self.assertEqual(list(self.stream), [b"line one\n", b"line two\n", b"line three\n"]) + self.assertEqual(self.buffer.to_attribute_value(), "line one\nline two\nline three\n") + + def test_readinto(self): + target = bytearray(8) + self.assertEqual(self.stream.readinto(target), 8) + self.assertEqual(target, b"line one") + self.assertEqual(self.buffer.to_attribute_value(), "line one") + + def test_seek_and_reread_is_not_duplicated(self): + self.stream.read(8) + self.assertEqual(self.stream.seek(0), 0) + self.stream.read() + self.assertEqual(self.buffer.to_attribute_value(), "line one\nline two\nline three\n") + + def test_seek_to_end_without_reading(self): + # Django checks the body size this way before reading it. + self.stream.seek(0, io.SEEK_END) + self.stream.seek(0) + self.stream.read(4) + self.assertEqual(self.buffer.to_attribute_value(), "line") + + def test_nothing_captured_without_reads(self): + self.assertIsNone(self.buffer.to_attribute_value()) + self.assertEqual(self.raw.tell(), 0) + + def test_optional_methods_only_present_when_wrapped_stream_has_them(self): + class ReadOnlyStream: + def __init__(self): + self._raw = io.BytesIO(b"body") + + def read(self, *args): + return self._raw.read(*args) + + def readline(self, *args): + return self._raw.readline(*args) + + def readlines(self, *args): + return self._raw.readlines(*args) + + def __iter__(self): + return iter(self._raw) + + stream = CapturingStream(ReadOnlyStream(), self.buffer) + self.assertFalse(hasattr(stream, "seek")) + self.assertFalse(hasattr(stream, "readinto")) + self.assertTrue(hasattr(self.stream, "seek")) + self.assertTrue(hasattr(self.stream, "readinto")) + self.assertEqual(stream.read(), b"body") + self.assertEqual(self.buffer.to_attribute_value(), "body") + + def test_other_attributes_are_forwarded(self): + self.assertTrue(self.stream.seekable()) + self.stream.read(4) + self.assertEqual(self.stream.tell(), 4) + self.stream.close() + self.assertTrue(self.raw.closed) + + def test_next(self): + self.assertEqual(next(self.stream), b"line one\n") + self.assertEqual(self.buffer.to_attribute_value(), "line one\n") + + def test_iteration_keeps_wrapped_stream_chunks(self): + class ChunkIterStream: + def __init__(self): + self._raw = io.BytesIO(b"hello\nworld\n") + + def read(self, *args): + return self._raw.read(*args) + + def readline(self, *args): + return self._raw.readline(*args) + + def __iter__(self): + return iter(lambda: self._raw.read(4), b"") + + stream = CapturingStream(ChunkIterStream(), self.buffer) + self.assertEqual(list(stream), [b"hell", b"o\nwo", b"rld\n"]) + self.assertEqual(self.buffer.to_attribute_value(), "hello\nworld\n") + + def test_unknown_method_call_stops_capture(self): + stream = CapturingStream(io.BufferedReader(io.BytesIO(b"hello\nworld\n")), self.buffer) + self.assertEqual(stream.read(2), b"he") + self.assertEqual(stream.read1(3), b"llo") + self.assertEqual(stream.read(), b"\nworld\n") + self.assertEqual(self.buffer.to_attribute_value(), "he") + + def test_unknown_method_lookup_keeps_capturing(self): + stream = CapturingStream(io.BufferedReader(io.BytesIO(b"hello\n")), self.buffer) + self.assertTrue(hasattr(stream, "read1")) + self.assertEqual(stream.read(), b"hello\n") + self.assertEqual(self.buffer.to_attribute_value(), "hello\n") + + def test_non_consuming_methods_keep_capturing(self): + stream = CapturingStream(io.BufferedReader(io.BytesIO(b"hello\n")), self.buffer) + self.assertTrue(stream.peek(1).startswith(b"h")) + self.assertEqual(stream.tell(), 0) + self.assertEqual(stream.read(), b"hello\n") + self.assertEqual(self.buffer.to_attribute_value(), "hello\n") + + def test_is_transparent_proxy(self): + self.assertIsInstance(self.stream, io.BytesIO) + with self.stream as entered: + self.assertIs(entered, self.raw) + self.assertTrue(self.raw.closed) + + def test_attribute_assignment_reaches_wrapped_stream(self): + self.stream.marker = "set" + self.assertEqual(vars(self.raw)["marker"], "set") diff --git a/uv.lock b/uv.lock index 9b9708d5980..6eda70d6e14 100644 --- a/uv.lock +++ b/uv.lock @@ -4230,6 +4230,12 @@ provides-extras = ["test", "upload"] [[package]] name = "opentelemetry-util-http" source = { editable = "util/opentelemetry-util-http" } +dependencies = [ + { name = "wrapt" }, +] + +[package.metadata] +requires-dist = [{ name = "wrapt", specifier = ">=1.0.0,<3.0.0" }] [[package]] name = "packaging"