Enforce size limits on browser SDK snippet injection to guard against decompression bombs and oversized HTML response bodies (#47233)

* Enforce size limits on browser SDK snippet injection to guard against decompression bombs and oversized HTML response bodies

* Add CHANGELOG

* Address feedback

* Fix format and lint

Author: rads-1996

sdk/monitor/azure-monitor-opentelemetry/CHANGELOG.md

@@ -7,6 +7,8 @@
 ### Breaking Changes
 
 ### Bugs Fixed
+- Enforce size limits on browser SDK snippet injection to guard against decompression bombs and oversized HTML response bodies (1 MiB compressed / 5 MiB decompressed caps).
+  ([#47233](https://github.com/Azure/azure-sdk-for-python/pull/47233))
 
 ### Other Changes
 

sdk/monitor/azure-monitor-opentelemetry/azure/monitor/opentelemetry/_browser_sdk_loader/snippet_injector.py

@@ -6,11 +6,16 @@
 
 import gzip
 import importlib
+import io
 import re
 from logging import getLogger
 from typing import Any, Dict, Optional, Tuple
 
 from ._config import BrowserSDKConfig
+from .._constants import (
+    _BROWSER_SDK_MAX_COMPRESSED_BYTES as _MAX_COMPRESSED_BYTES,
+    _BROWSER_SDK_MAX_DECOMPRESSED_BYTES as _MAX_DECOMPRESSED_BYTES,
+)
 
 # Optional compression libraries
 _BROTLI_MODULE: Optional[Any]
@@ -34,6 +39,58 @@
 _logger = getLogger(__name__)
 
 
+def _bounded_decompress(data: bytes, encoding: str) -> bytes:
+    """Decompress ``data`` with the given encoding, raising if output exceeds the cap.
+
+    Supports ``gzip``, ``deflate`` (zlib), and ``br`` (brotli). Returns the
+    decompressed bytes or raises ``ValueError`` if the output would exceed
+    ``_MAX_DECOMPRESSED_BYTES`` (defense against decompression bombs).
+
+    :param data: Compressed input bytes.
+    :type data: bytes
+    :param encoding: One of ``gzip``, ``deflate``, ``br``.
+    :type encoding: str
+    :return: Decompressed bytes (at most ``_MAX_DECOMPRESSED_BYTES``).
+    :rtype: bytes
+    """
+    cap = _MAX_DECOMPRESSED_BYTES
+    if encoding == "gzip":
+        # GzipFile.read(N) reads at most N bytes; ask for cap+1 to detect overflow.
+        with gzip.GzipFile(fileobj=io.BytesIO(data)) as gz:
+            out = gz.read(cap + 1)
+    elif encoding == "deflate":
+        if _ZLIB_MODULE is None:
+            raise RuntimeError("zlib module not available")
+        # zlib's decompressobj supports a native max_length parameter.
+        decompressor = _ZLIB_MODULE.decompressobj()
+        out = decompressor.decompress(data, cap + 1)
+        if decompressor.unconsumed_tail or len(out) > cap:
+            raise ValueError("deflate-decompressed body exceeds cap")
+        out += decompressor.flush()
+        if len(out) > cap:
+            raise ValueError("deflate-decompressed body exceeds cap")
+        return bytes(out)
+    elif encoding == "br":
+        if _BROTLI_MODULE is None:
+            raise RuntimeError("brotli module not available")
+        # Brotli has no built-in size limit; feed chunks and tally bytes.
+        decompressor = _BROTLI_MODULE.Decompressor()
+        chunk = 64 * 1024
+        pieces, total = [], 0
+        for start in range(0, len(data), chunk):
+            piece = decompressor.process(data[start : start + chunk])
+            total += len(piece)
+            if total > cap:
+                raise ValueError("brotli-decompressed body exceeds cap")
+            pieces.append(piece)
+        return b"".join(pieces)
+    else:
+        raise ValueError(f"unsupported encoding: {encoding}")
+    if len(out) > cap:
+        raise ValueError(f"{encoding}-decompressed body exceeds cap")
+    return out
+
+
 def _mark_browser_loader_feature(is_enabled: bool) -> None:
     """Record browser SDK loader usage in statsbeat when available.
 
@@ -107,7 +164,7 @@ def __init__(self, config: BrowserSDKConfig) -> None:
         ]
         _mark_browser_loader_feature(self.config.enabled)
 
-    def should_inject(
+    def should_inject(  # pylint: disable=too-many-return-statements
         self, request_method: str, content_type: Optional[str], content: bytes, content_encoding: Optional[str] = None
     ) -> bool:
         """Determine whether the web snippet should be injected into the response.
@@ -131,8 +188,21 @@ def should_inject(
         # Check content type for HTML
         if not content_type or "html" not in content_type.lower():
             return False
-        # Get decompressed content once and cache it for reuse
-        decompressed_content = self._get_decompressed_content(content, content_encoding)
+        # Bail out on oversized bodies; never decompress/scan/recompress them.
+        if len(content) > _MAX_COMPRESSED_BYTES:
+            _logger.debug(
+                "Response body %d bytes exceeds injection cap; skipping snippet injection",
+                len(content),
+            )
+            return False
+        # Get decompressed content once and cache it for reuse. If decompression
+        # fails (e.g., size cap exceeded or malformed body), skip injection rather
+        # than risk modifying a body we cannot safely decode.
+        try:
+            decompressed_content = self._get_decompressed_content(content, content_encoding)
+        except Exception as ex:  # pylint: disable=broad-exception-caught
+            _logger.debug("Skipping snippet injection; decompression failed: %s", ex)
+            return False
         # Check if Web SDK is already present using cached decompressed content
         if self._has_existing_web_sdk_from_decompressed(decompressed_content):
             _logger.debug("Web SDK already detected in HTML, skipping injection")
@@ -372,19 +442,23 @@ def _decompress_content(self, content: bytes, encoding: Optional[str]) -> bytes:
         try:
             normalized = encoding.lower()
             if normalized == "gzip":
-                result = gzip.decompress(content)
+                result = _bounded_decompress(content, "gzip")
             elif normalized == "br":
                 if not HAS_BROTLI or _BROTLI_MODULE is None:
                     _logger.warning("brotli library not available for decompression")
                 else:
-                    result = _BROTLI_MODULE.decompress(content)
+                    result = _bounded_decompress(content, "br")
             elif normalized == "deflate":
                 if not HAS_ZLIB or _ZLIB_MODULE is None:
                     _logger.warning("zlib library not available for decompression")
                 else:
-                    result = _ZLIB_MODULE.decompress(content)
+                    result = _bounded_decompress(content, "deflate")
         except Exception as ex:  # pylint: disable=broad-exception-caught
+            # Re-raise so callers (e.g., inject_with_compression / should_inject) can
+            # fall back to returning the original response untouched instead of
+            # treating still-compressed bytes as HTML and double-compressing them.
             _logger.warning("Failed to decompress content with encoding %s: %s", encoding, ex)
+            raise
         return result
 
     def _compress_content(self, content: bytes, encoding: str) -> bytes:

sdk/monitor/azure-monitor-opentelemetry/azure/monitor/opentelemetry/_constants.py

@@ -89,3 +89,8 @@
 
 _AZURE_APP_SERVICE_RESOURCE_DETECTOR_NAME = "azure_app_service"
 _AZURE_VM_RESOURCE_DETECTOR_NAME = "azure_vm"
+
+# --------------------Browser SDK snippet injection------------------------------
+
+_BROWSER_SDK_MAX_COMPRESSED_BYTES = 1 * 1024 * 1024  # 1 MiB on the wire
+_BROWSER_SDK_MAX_DECOMPRESSED_BYTES = 5 * 1024 * 1024  # 5 MiB after decompression

sdk/monitor/azure-monitor-opentelemetry/tests/browserSdkLoader/test_snippet_injector.py

@@ -6,10 +6,17 @@
 
 import gzip
 import unittest
+import zlib
 from unittest.mock import patch, MagicMock
 
 from azure.monitor.opentelemetry._browser_sdk_loader._config import BrowserSDKConfig
-from azure.monitor.opentelemetry._browser_sdk_loader.snippet_injector import WebSnippetInjector
+from azure.monitor.opentelemetry._browser_sdk_loader import snippet_injector as snippet_injector_module
+from azure.monitor.opentelemetry._browser_sdk_loader.snippet_injector import (
+    WebSnippetInjector,
+    _bounded_decompress,
+    _MAX_COMPRESSED_BYTES,
+    _MAX_DECOMPRESSED_BYTES,
+)
 
 
 class TestWebSnippetInjector(unittest.TestCase):
@@ -371,5 +378,77 @@ def test_should_inject_script_url_detection(self):
                 self.assertFalse(result, f"Should detect existing script URL in: {content}")
 
 
+class TestSizeCaps(unittest.TestCase):
+    """Tests for the response-size and decompression-size guards."""
+
+    def setUp(self):
+        self.config = BrowserSDKConfig(
+            enabled=True,
+            connection_string=(
+                "InstrumentationKey=12345678-1234-1234-1234-123456789012;"
+                "IngestionEndpoint=https://test.in.applicationinsights.azure.com/"
+            ),
+        )
+        self.injector = WebSnippetInjector(self.config)
+
+    def test_should_inject_rejects_oversized_body(self):
+        """Bodies larger than the compressed cap must be skipped before any decompression."""
+        oversized = b"<html><body>" + b"a" * (_MAX_COMPRESSED_BYTES + 1) + b"</body></html>"
+        # Patch decompression to make sure we don't even reach it.
+        with patch.object(self.injector, "_get_decompressed_content") as mock_decompress:
+            result = self.injector.should_inject("GET", "text/html", oversized)
+        self.assertFalse(result)
+        mock_decompress.assert_not_called()
+
+    def test_should_inject_accepts_body_at_cap(self):
+        """A body exactly at the cap should still be processed."""
+        prefix = b"<html><head></head><body>"
+        suffix = b"</body></html>"
+        body = prefix + (b"a" * (_MAX_COMPRESSED_BYTES - len(prefix) - len(suffix))) + suffix
+        self.assertEqual(len(body), _MAX_COMPRESSED_BYTES)
+        self.assertTrue(self.injector.should_inject("GET", "text/html", body))
+
+    def test_bounded_decompress_gzip_under_cap(self):
+        payload = b"<html>hello</html>"
+        self.assertEqual(_bounded_decompress(gzip.compress(payload), "gzip"), payload)
+
+    def test_bounded_decompress_gzip_bomb_raises(self):
+        bomb = gzip.compress(b"a" * (_MAX_DECOMPRESSED_BYTES + 1024))
+        with self.assertRaises(ValueError):
+            _bounded_decompress(bomb, "gzip")
+
+    def test_bounded_decompress_deflate_under_cap(self):
+        payload = b"<html>hello</html>"
+        self.assertEqual(_bounded_decompress(zlib.compress(payload), "deflate"), payload)
+
+    def test_bounded_decompress_deflate_bomb_raises(self):
+        bomb = zlib.compress(b"a" * (_MAX_DECOMPRESSED_BYTES + 1024))
+        with self.assertRaises(ValueError):
+            _bounded_decompress(bomb, "deflate")
+
+    def test_bounded_decompress_brotli_under_cap(self):
+        if not snippet_injector_module.HAS_BROTLI:
+            self.skipTest("brotli not installed")
+        payload = b"<html>hello</html>"
+        compressed = snippet_injector_module._BROTLI_MODULE.compress(payload)
+        self.assertEqual(_bounded_decompress(compressed, "br"), payload)
+
+    def test_bounded_decompress_brotli_bomb_raises(self):
+        if not snippet_injector_module.HAS_BROTLI:
+            self.skipTest("brotli not installed")
+        bomb = snippet_injector_module._BROTLI_MODULE.compress(b"a" * (_MAX_DECOMPRESSED_BYTES + 1024))
+        with self.assertRaises(ValueError):
+            _bounded_decompress(bomb, "br")
+
+    def test_inject_with_compression_returns_original_on_bomb(self):
+        """A gzip bomb must not be expanded into memory by inject_with_compression."""
+        bomb = gzip.compress(b"a" * (_MAX_DECOMPRESSED_BYTES + 1024))
+        modified, encoding = self.injector.inject_with_compression(bomb, "gzip")
+        # On decompression cap / failure, injection must be skipped and the
+        # original body returned unchanged (no double-compression).
+        self.assertEqual(modified, bomb)
+        self.assertEqual(encoding, "gzip")
+
+
 if __name__ == "__main__":
     unittest.main()