fix(security): neutralize CR/LF in rendered log records (CWE-117)

src/youtube_extension/backend/config/logging_config.py

@@ -14,6 +14,25 @@
 from datetime import datetime
 from pathlib import Path
 
+# CWE-117: characters that let attacker-controlled text forge or corrupt log
+# lines. CR/LF are the classic vector; VT/FF and the Unicode line/paragraph
+# separators (plus NEL) are treated as line breaks by some log processors, and
+# ESC enables terminal-escape injection. We neutralize the FINAL rendered record
+# so that no sink -- including exc_info tracebacks and structured `extra` fields
+# that never pass through an inline sanitizer -- can inject a physical log line.
+# Escaping (rather than dropping) keeps multi-line tracebacks fully diagnosable
+# on a single physical line with zero information loss.
+_UNSAFE_LOG_CHARS = {
+    ord("\r"): "\\r",
+    ord("\n"): "\\n",
+    ord("\v"): "\\v",
+    ord("\f"): "\\f",
+    ord("\x1b"): "\\x1b",
+    ord("\x85"): "\\x85",
+    ord("\u2028"): "\\u2028",
+    ord("\u2029"): "\\u2029",
+}
+
 
 class StructuredFormatter(logging.Formatter):
     """
@@ -36,10 +55,15 @@ def format(self, record: logging.LogRecord) -> str:
         if hasattr(record, 'request_id'):
             record.correlation_id = record.request_id
 
-        # Format the base message
+        # Format the base message (this also appends any exc_info traceback
+        # and stack_info that the base formatter renders).
         formatted_message = super().format(record)
 
-        return formatted_message
+        # CWE-117: neutralize line/escape separators in the fully rendered
+        # record so that message, traceback, and structured extras can never
+        # forge a log line -- even when the caller did not sanitize inputs at
+        # the call site.
+        return formatted_message.translate(_UNSAFE_LOG_CHARS)
 
     def formatException(self, ei) -> str:
         """Format exception with enhanced stack trace"""

src/youtube_extension/backend/services/intelligent_cache.py

@@ -286,7 +286,12 @@ def __init__(self, name: str = "L2_Redis", redis_url: str = "redis://localhost:6
         self._tag_write_semaphore_loop: Optional[asyncio.AbstractEventLoop] = None
 
     def _get_tag_write_semaphore(self) -> asyncio.Semaphore:
-        """Semaphore shared by every ``set()`` call on this layer.
+        """Semaphore shared by every tag fan-out on this layer.
+
+        Two paths acquire it: ``set()``, which issues one ``sadd`` per tag, and
+        ``invalidate_by_tags()``, which issues an ``smembers``/``delete`` pair
+        per tag. Both draw from the same budget, so a ``set()`` storm and an
+        invalidation storm cannot each claim ``_tag_write_limit`` connections.
 
         The limiter has to be per-instance rather than per-call: all callers
         share ``self.redis_pool``, so a per-call semaphore would let N
@@ -310,11 +315,12 @@ def _get_tag_write_semaphore(self) -> asyncio.Semaphore:
         ``redis.asyncio`` pool caches connections whose transports are bound to
         the loop that opened them, so a ``RedisCacheLayer`` is already
         event-loop-affine through ``self.redis_pool`` -- and that affinity
-        applies equally to ``get()``, ``delete()``, ``clear()`` and
-        ``invalidate_by_tags()``, none of which this limiter touches. Enforcing
-        a loop-ownership contract is a layer-wide concern tracked in #1162;
-        guarding only this one path would give a misleading partial guarantee.
-        Use one layer per event loop.
+        applies equally to ``get()``, ``delete()`` and ``clear()``, none of
+        which this limiter touches -- and to the two paths that do acquire it,
+        since bounding fan-out is not the same guarantee as owning a loop.
+        Enforcing a loop-ownership contract is a layer-wide concern tracked in
+        #1162; guarding only these paths would give a misleading partial
+        guarantee. Use one layer per event loop.
         """
         loop = asyncio.get_running_loop()
 
@@ -504,19 +510,42 @@ async def invalidate_by_tags(self, tags: list[str]) -> int:
 
         try:
             async with redis.Redis(connection_pool=self.redis_pool) as conn:
-                total_deleted = 0
+                semaphore = self._get_tag_write_semaphore()
 
-                for tag in tags:
-                    # Get all keys with this tag
-                    keys = await conn.smembers(f"uvai:tag:{tag}")
+                async def _invalidate_tag(tag: str) -> int:
+                    # One permit covers both commands for a tag rather than one
+                    # each. The delete operates on the members smembers just
+                    # returned, so the pair is causally ordered and cannot be
+                    # interleaved; holding the permit across both keeps the
+                    # number of concurrently held pool connections equal to the
+                    # permit count instead of twice it.
+                    async with semaphore:
+                        keys = await conn.smembers(f"uvai:tag:{tag}")
+
+                        if not keys:
+                            return 0
 
-                    if keys:
                         # Delete cache entries
                         cache_keys = [f"uvai:cache:{key.decode()}" if isinstance(key, bytes) else f"uvai:cache:{key}" for key in keys]
                         stat_keys = [f"uvai:stats:{key.decode()}" if isinstance(key, bytes) else f"uvai:stats:{key}" for key in keys]
 
-                        deleted = await conn.delete(*(cache_keys + stat_keys + [f"uvai:tag:{tag}"]))
-                        total_deleted += deleted
+                        return await conn.delete(*(cache_keys + stat_keys + [f"uvai:tag:{tag}"]))
+
+                # return_exceptions=True so that one failing tag cannot leave
+                # sibling tasks still in flight once this method returns, which
+                # would let them touch conn after the enclosing async with has
+                # closed it. The first failure is re-raised below so the
+                # existing handler still reports 0.
+                results = await asyncio.gather(
+                    *(_invalidate_tag(tag) for tag in tags),
+                    return_exceptions=True,
+                )
+
+                total_deleted = 0
+                for result in results:
+                    if isinstance(result, BaseException):
+                        raise result
+                    total_deleted += result
 
                 logger.info(f"L2 Redis TAG INVALIDATION: {total_deleted} entries for tags {tags}")
                 return total_deleted

tests/unit/test_intelligent_cache.py

@@ -1593,6 +1593,164 @@ async def test_invalidate_exception_returns_zero(self):
 
         assert result == 0
 
+    async def test_invalidate_issues_tags_concurrently(self):
+        """Per-tag work must overlap rather than run one tag at a time.
+
+        This is the non-vacuity guard for the change: a serial ``for`` loop
+        yields a peak of exactly 1, so this assertion fails on the previous
+        implementation.
+        """
+        layer = self._connected_layer()
+        conn = _make_redis_conn()
+
+        in_flight = 0
+        peak = 0
+
+        async def _tracking_smembers(*_args, **_kwargs):
+            nonlocal in_flight, peak
+            in_flight += 1
+            peak = max(peak, in_flight)
+            # Yield so sibling tags can start if the fan-out is concurrent.
+            await asyncio.sleep(0)
+            in_flight -= 1
+            return {b"key1"}
+
+        conn.smembers = AsyncMock(side_effect=_tracking_smembers)
+        conn.delete = AsyncMock(return_value=1)
+
+        with _patch_redis(conn):
+            result = await layer.invalidate_by_tags([f"tag{i}" for i in range(5)])
+
+        assert peak > 1
+        assert conn.smembers.call_count == 5
+        assert result == 5
+
+    async def test_invalidate_stays_within_concurrency_bound(self):
+        """A large tag list must not fan out past the connection-pool budget."""
+        layer = self._connected_layer()
+        conn = _make_redis_conn()
+
+        in_flight = 0
+        peak = 0
+
+        async def _tracking_smembers(*_args, **_kwargs):
+            nonlocal in_flight, peak
+            in_flight += 1
+            peak = max(peak, in_flight)
+            await asyncio.sleep(0)
+            in_flight -= 1
+            return set()
+
+        conn.smembers = AsyncMock(side_effect=_tracking_smembers)
+
+        with _patch_redis(conn):
+            result = await layer.invalidate_by_tags([f"tag{i}" for i in range(50)])
+
+        assert result == 0
+        assert conn.smembers.call_count == 50
+        assert peak <= layer._tag_write_limit
+
+    async def test_invalidate_holds_one_permit_across_both_commands(self):
+        """smembers and delete for a tag must not be split across permits.
+
+        The delete operates on the members smembers just returned, so a permit
+        that is released between them would let the number of concurrently held
+        pool connections reach twice the budget.
+        """
+        layer = self._connected_layer()
+        layer._tag_write_limit = 1
+        conn = _make_redis_conn()
+
+        order = []
+
+        async def _smembers(name, *_args, **_kwargs):
+            order.append(("smembers", name))
+            await asyncio.sleep(0)
+            return {b"key1"}
+
+        async def _delete(*args, **_kwargs):
+            order.append(("delete", args[-1]))
+            await asyncio.sleep(0)
+            return 1
+
+        conn.smembers = AsyncMock(side_effect=_smembers)
+        conn.delete = AsyncMock(side_effect=_delete)
+
+        with _patch_redis(conn):
+            await layer.invalidate_by_tags(["tag1", "tag2"])
+
+        # With one permit the pairs must not interleave.
+        assert order == [
+            ("smembers", "uvai:tag:tag1"),
+            ("delete", "uvai:tag:tag1"),
+            ("smembers", "uvai:tag:tag2"),
+            ("delete", "uvai:tag:tag2"),
+        ]
+
+    async def test_invalidate_failure_drains_in_flight_work(self):
+        """A failing tag returns 0 with no per-tag task left in flight."""
+        layer = self._connected_layer()
+        conn = _make_redis_conn()
+
+        started = 0
+        finished = 0
+
+        async def _flaky_smembers(name, *_args, **_kwargs):
+            nonlocal started, finished
+            started += 1
+            await asyncio.sleep(0)
+            finished += 1
+            if name.endswith("tag3"):
+                raise RuntimeError("redis unavailable")
+            return set()
+
+        conn.smembers = AsyncMock(side_effect=_flaky_smembers)
+
+        with _patch_redis(conn):
+            result = await layer.invalidate_by_tags([f"tag{i}" for i in range(6)])
+
+        assert result == 0
+        # Every scheduled tag ran to completion before the method returned.
+        assert started == 6
+        assert finished == started
+
+    async def test_invalidate_shares_tag_write_budget_with_set(self):
+        """set() and invalidate_by_tags() must draw from one shared budget.
+
+        Both hold connections from the same pool, so separate budgets would let
+        a concurrent set storm and invalidation storm each claim the full limit.
+        """
+        layer = self._connected_layer()
+        conn = _make_redis_conn()
+
+        in_flight = 0
+        peak = 0
+
+        async def _tracked(*_args, **_kwargs):
+            nonlocal in_flight, peak
+            in_flight += 1
+            peak = max(peak, in_flight)
+            await asyncio.sleep(0)
+            in_flight -= 1
+            return 1
+
+        async def _tracked_smembers(*_args, **_kwargs):
+            await _tracked()
+            return set()
+
+        conn.sadd = AsyncMock(side_effect=_tracked)
+        conn.smembers = AsyncMock(side_effect=_tracked_smembers)
+
+        with _patch_redis(conn):
+            await asyncio.gather(
+                layer.set("k", "v", tags=[f"s{i}" for i in range(40)]),
+                layer.invalidate_by_tags([f"i{i}" for i in range(40)]),
+            )
+
+        assert conn.sadd.call_count == 40
+        assert conn.smembers.call_count == 40
+        assert peak <= layer._tag_write_limit
+
 
 class TestRedisCacheLayerUpdateAvgAccessTime:
     """Tests for RedisCacheLayer._update_avg_access_time() — lines 442-450"""

tests/unit/test_logging_config_crlf.py

@@ -0,0 +1,100 @@
+"""
+Regression tests for CWE-117 (log injection / log forging) hardening in
+``StructuredFormatter``.
+
+These assert against the *rendered* handler output — not the return value of any
+inline sanitizer — so they cover every sink the formatter touches, including
+``exc_info`` tracebacks and structured ``extra`` fields that never pass through
+a call-site sanitizer.
+"""
+
+import io
+import logging
+
+import pytest
+
+from youtube_extension.backend.config.logging_config import (
+    _UNSAFE_LOG_CHARS,
+    StructuredFormatter,
+)
+
+
+def _render(record_emitter) -> str:
+    """Emit one or more records through a StructuredFormatter and return output."""
+    buf = io.StringIO()
+    handler = logging.StreamHandler(buf)
+    handler.setFormatter(StructuredFormatter("%(levelname)s - %(message)s"))
+    logger = logging.getLogger("crlf-regression")
+    logger.handlers[:] = [handler]
+    logger.propagate = False
+    logger.setLevel(logging.INFO)
+    record_emitter(logger)
+    return buf.getvalue()
+
+
+def _forged_lines(output: str) -> list:
+    """Physical lines that would appear as their own forged log entries."""
+    return [line for line in output.split("\n") if line.startswith("CRITICAL - FORGED")]
+
+
+pytestmark = [pytest.mark.unit, pytest.mark.security]
+
+
+def test_message_crlf_cannot_forge_log_lines():
+    out = _render(
+        lambda lg: lg.info("video_id=%s", "abc\r\nCRITICAL - FORGED VIA MESSAGE")
+    )
+    assert "\r" not in out
+    assert _forged_lines(out) == []
+    # The literal payload is preserved (escaped), so nothing is silently lost.
+    assert "CRITICAL - FORGED VIA MESSAGE" in out
+
+
+def test_exc_info_traceback_cannot_forge_log_lines():
+    def emit(lg):
+        try:
+            raise ValueError("boom\r\nCRITICAL - FORGED VIA EXC")
+        except ValueError:
+            lg.error("Error in chat endpoint", exc_info=True)
+
+    out = _render(emit)
+    # A single ``lg.error(..., exc_info=True)`` must render as exactly one
+    # physical line no matter how many newlines the traceback contains.
+    physical = [line for line in out.split("\n") if line]
+    assert len(physical) == 1
+    assert "\r" not in out
+    assert _forged_lines(out) == []
+    # Traceback content is still present (escaped) for diagnosability.
+    assert "Traceback (most recent call last)" in out
+    assert "ValueError: boom" in out
+
+
+def test_structured_extra_and_unicode_separators_cannot_forge_log_lines():
+    ls = chr(0x2028)  # Unicode LINE SEPARATOR
+    out = _render(
+        lambda lg: lg.info(
+            "url=%s",
+            "http://x" + ls + "CRITICAL - FORGED VIA LS",
+            extra={"request_id": "r\n1"},
+        )
+    )
+    assert ls not in out
+    assert "\r" not in out
+    assert "\n" not in out.rstrip("\n")
+    assert _forged_lines(out) == []
+
+
+@pytest.mark.parametrize("char", sorted(_UNSAFE_LOG_CHARS))
+def test_every_declared_unsafe_char_is_neutralized(char):
+    payload = "before" + chr(char) + "after"
+    out = _render(lambda lg: lg.info("v=%s", payload))
+    # Strip only the handler's own trailing line terminator before inspecting
+    # the record body — for char == "\n" that terminator is the sole legitimate
+    # newline in the stream.
+    body = out.rstrip("\n")
+    # The raw separator/control character must not survive into the record body.
+    assert chr(char) not in body
+    # Its escaped form must appear instead.
+    assert _UNSAFE_LOG_CHARS[char] in body
+    # The record must remain a single physical line.
+    assert body.count("\n") == 0