From 7aababa63d51c0c7b4f0d2cb294194164db5de4d Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Sun, 2 Aug 2026 13:20:20 -0500 Subject: [PATCH 1/2] perf: invalidate Redis tags concurrently instead of serially invalidate_by_tags() issued smembers+delete one tag at a time, so the caller's wall-clock latency grew linearly with tag count and stale entries stayed readable for the whole window. Fan the per-tag work out with asyncio.gather, reusing the per-layer tag-write semaphore already introduced for set() so the combined fan-out cannot exhaust the shared connection pool. One permit covers both commands for a tag since the delete depends on the smembers result. Closes #1261. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../backend/services/intelligent_cache.py | 55 ++++-- tests/unit/test_intelligent_cache.py | 158 ++++++++++++++++++ 2 files changed, 200 insertions(+), 13 deletions(-) diff --git a/src/youtube_extension/backend/services/intelligent_cache.py b/src/youtube_extension/backend/services/intelligent_cache.py index a0281de08..34e9f0694 100644 --- a/src/youtube_extension/backend/services/intelligent_cache.py +++ b/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 diff --git a/tests/unit/test_intelligent_cache.py b/tests/unit/test_intelligent_cache.py index 591cb0c12..db55b0468 100644 --- a/tests/unit/test_intelligent_cache.py +++ b/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""" From 97863c1a147ae07a704c95ae8f3f2df3c8de1de0 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 2 Aug 2026 18:30:06 +0000 Subject: [PATCH 2/2] fix(security): neutralize CR/LF in rendered log records (CWE-117) StructuredFormatter previously returned the fully rendered record verbatim, so any newline/carriage-return in the record body could forge additional log lines. Inline call-site sanitizers cannot cover every sink: exc_info tracebacks (str(exc) + frames) and structured `extra` fields are appended by the formatter itself, after any per-argument sanitization runs. Neutralize line/escape separators (CR, LF, VT, FF, ESC, NEL, U+2028, U+2029) in the final rendered string inside StructuredFormatter.format(), so message, traceback, and extras are all covered at a single choke point regardless of the call site. Separators are escaped (not dropped), keeping multi-line tracebacks fully diagnosable on one physical line with no information loss. Adds tests/unit/test_logging_config_crlf.py asserting against rendered handler output for the message, exc_info-traceback, structured-extra, and Unicode-separator vectors, plus a parametrized check that every declared unsafe character is neutralized. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01RirdQCKDByp2a8qdLb8PXh --- .../backend/config/logging_config.py | 28 ++++- tests/unit/test_logging_config_crlf.py | 100 ++++++++++++++++++ 2 files changed, 126 insertions(+), 2 deletions(-) create mode 100644 tests/unit/test_logging_config_crlf.py diff --git a/src/youtube_extension/backend/config/logging_config.py b/src/youtube_extension/backend/config/logging_config.py index 98f28489e..99f73884a 100644 --- a/src/youtube_extension/backend/config/logging_config.py +++ b/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""" diff --git a/tests/unit/test_logging_config_crlf.py b/tests/unit/test_logging_config_crlf.py new file mode 100644 index 000000000..1c3307803 --- /dev/null +++ b/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