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/4] 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 e0ebdb3697317c4d8739d62d8ce3755bc5d036bd Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 2 Aug 2026 18:34:31 +0000 Subject: [PATCH 2/4] docs(cache): correct permit-scope rationale + add cancellation guard CodeRabbit's adversarial review of #1262 found the permit-scope comment factually wrong: it claimed holding one permit across the smembers/delete pair keeps concurrently-held pool connections at the permit count "instead of twice it". redis.asyncio checks a connection out only for each command and returns it to the pool between the two awaits, so acquiring the permit per command would cap in-flight commands at the same limit. The real reason to hold across the pair is scheduling determinism and avoiding permit churn, not preventing a doubling of held connections. - Rewrite the inline comment in invalidate_by_tags() to state the true policy. - Fix the matching docstring of test_invalidate_holds_one_permit_across_both_commands. - Add test_invalidate_starts_no_command_after_context_exit_on_cancel, the cancellation regression guard CodeRabbit asked for: cancel mid-smembers and assert no Redis command starts after the connection context manager exits. No behaviour change; scheduling and return values are untouched. 163 passed. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01RJP3fieq6JJdEAZXQsJ3J6 --- .../backend/services/intelligent_cache.py | 20 ++++-- tests/unit/test_intelligent_cache.py | 68 +++++++++++++++++-- 2 files changed, 77 insertions(+), 11 deletions(-) diff --git a/src/youtube_extension/backend/services/intelligent_cache.py b/src/youtube_extension/backend/services/intelligent_cache.py index 34e9f0694..616ee39a7 100644 --- a/src/youtube_extension/backend/services/intelligent_cache.py +++ b/src/youtube_extension/backend/services/intelligent_cache.py @@ -513,12 +513,20 @@ async def invalidate_by_tags(self, tags: list[str]) -> int: semaphore = self._get_tag_write_semaphore() 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. + # One permit is held for the whole smembers->delete pair + # rather than re-acquired per command. This bounds the + # number of tag invalidations in progress at once to the + # permit count and keeps each tag's causally-ordered pair + # (delete operates on the members smembers just returned) as + # one indivisible unit of scheduled work. + # + # It does NOT lower peak pool-connection usage: redis.asyncio + # checks a connection out only for the duration of each + # command and returns it to the pool between the two awaits, + # so acquiring the permit per command would cap in-flight + # commands at the same limit. The reason to hold across the + # pair is scheduling determinism and avoiding permit churn, + # not preventing a doubling of held connections. async with semaphore: keys = await conn.smembers(f"uvai:tag:{tag}") diff --git a/tests/unit/test_intelligent_cache.py b/tests/unit/test_intelligent_cache.py index db55b0468..26c7ccfbc 100644 --- a/tests/unit/test_intelligent_cache.py +++ b/tests/unit/test_intelligent_cache.py @@ -1651,11 +1651,14 @@ async def _tracking_smembers(*_args, **_kwargs): 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. + """smembers and delete for a tag are scheduled under one permit. + + Holding the permit across the pair keeps each tag's invalidation as one + indivisible unit of scheduled work, so with a single permit the pairs + run to completion without interleaving. (This does not change peak pool + usage -- redis.asyncio returns a connection to the pool between the two + awaits -- it pins the per-tag scheduling policy so a later refactor + cannot silently split the pair.) """ layer = self._connected_layer() layer._tag_write_limit = 1 @@ -1687,6 +1690,61 @@ async def _delete(*args, **_kwargs): ("delete", "uvai:tag:tag2"), ] + async def test_invalidate_starts_no_command_after_context_exit_on_cancel(self): + """A cancelled invalidate issues no Redis command after conn closes. + + Regression guard for the cancellation-parity claim: if the task running + invalidate_by_tags() is cancelled while a child is blocked in smembers, + gather cancels the children and CancelledError unwinds out through the + ``async with redis.Redis(...)`` context manager. A cancelled child may + still run its ``finally`` (releasing the semaphore), but it must never + start a new command on the connection once __aexit__ has closed it. + """ + layer = self._connected_layer() + layer._tag_write_limit = 2 + conn = _make_redis_conn() + + events: list[tuple[str, object]] = [] + first_smembers = asyncio.Event() + + async def _blocking_smembers(name, *_args, **_kwargs): + events.append(("smembers_start", name)) + first_smembers.set() + await asyncio.sleep(3600) # block until cancelled + events.append(("smembers_end", name)) # unreachable once cancelled + return set() + + async def _delete(*args, **_kwargs): + events.append(("delete_start", args[-1])) + return 1 + + conn.smembers = AsyncMock(side_effect=_blocking_smembers) + conn.delete = AsyncMock(side_effect=_delete) + + original_aexit = conn.__aexit__ + + async def _tracking_aexit(*args, **kwargs): + events.append(("aexit", None)) + return await original_aexit(*args, **kwargs) + + conn.__aexit__ = AsyncMock(side_effect=_tracking_aexit) + + with _patch_redis(conn): + task = asyncio.ensure_future(layer.invalidate_by_tags(["t1", "t2"])) + await first_smembers.wait() # a child is now mid-smembers + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + # The connection context manager exited, and nothing issued a command + # after it did. + assert ("aexit", None) in events + aexit_index = events.index(("aexit", None)) + after_exit = events[aexit_index + 1 :] + assert not any( + kind in ("smembers_start", "delete_start") for kind, _ in after_exit + ) + 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() From 968b5b1c904638b68d43ee846c726ac4d8d02d48 Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Sun, 2 Aug 2026 13:38:02 -0500 Subject: [PATCH 3/4] test(cache): replace vacuous cancel guard with prove-failing drain test The cancellation guard added in e0ebdb369 passes against the pre-change serial implementation on main, so it cannot detect a regression back to sequential invalidation, and its smembers mock has no try/finally, so it never observes whether a cancelled child actually unwound. Measured: running it against `git show origin/main:intelligent_cache.py` yields 1 passed. Replace it with test_invalidate_cancellation_drains_before_conn_closes, which asserts a strict superset: - all three tags are in flight concurrently, via a bounded wait that fails with a diagnostic instead of hanging when they are not; - every cancelled child runs its finally before __aexit__ closes the connection (recorded by a try/finally in the smembers mock); - no Redis command starts after __aexit__ (the original assertion). Against pristine main this test fails, taking the class prove-fail count from 3 to 4. Also notes in the permit-scope comment that holding one permit across the pair bounds how many tags can sit half-invalidated when a delete fails. Full file 163 passed; wider sweep 474 passed; ruff parity unchanged. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../backend/services/intelligent_cache.py | 3 +- tests/unit/test_intelligent_cache.py | 89 +++++++++++-------- 2 files changed, 55 insertions(+), 37 deletions(-) diff --git a/src/youtube_extension/backend/services/intelligent_cache.py b/src/youtube_extension/backend/services/intelligent_cache.py index 616ee39a7..a92cbe171 100644 --- a/src/youtube_extension/backend/services/intelligent_cache.py +++ b/src/youtube_extension/backend/services/intelligent_cache.py @@ -518,7 +518,8 @@ async def _invalidate_tag(tag: str) -> int: # number of tag invalidations in progress at once to the # permit count and keeps each tag's causally-ordered pair # (delete operates on the members smembers just returned) as - # one indivisible unit of scheduled work. + # one indivisible unit of scheduled work. It also bounds how + # many tags can sit half-invalidated if a delete fails. # # It does NOT lower peak pool-connection usage: redis.asyncio # checks a connection out only for the duration of each diff --git a/tests/unit/test_intelligent_cache.py b/tests/unit/test_intelligent_cache.py index 26c7ccfbc..b59b3be50 100644 --- a/tests/unit/test_intelligent_cache.py +++ b/tests/unit/test_intelligent_cache.py @@ -1690,60 +1690,77 @@ async def _delete(*args, **_kwargs): ("delete", "uvai:tag:tag2"), ] - async def test_invalidate_starts_no_command_after_context_exit_on_cancel(self): - """A cancelled invalidate issues no Redis command after conn closes. - - Regression guard for the cancellation-parity claim: if the task running - invalidate_by_tags() is cancelled while a child is blocked in smembers, - gather cancels the children and CancelledError unwinds out through the - ``async with redis.Redis(...)`` context manager. A cancelled child may - still run its ``finally`` (releasing the semaphore), but it must never - start a new command on the connection once __aexit__ has closed it. + async def test_invalidate_cancellation_drains_before_conn_closes(self): + """Cancellation must unwind every child before the connection closes. + + This is the explicit cancellation-parity claim: gather() does not + complete its outer future until every cancelled child has finished, so + the enclosing ``async with redis.Redis(...)`` cannot close ``conn`` + while a child could still issue a command on it. """ layer = self._connected_layer() - layer._tag_write_limit = 2 + layer._tag_write_limit = 4 conn = _make_redis_conn() - events: list[tuple[str, object]] = [] - first_smembers = asyncio.Event() + events: list[tuple] = [] + all_blocked = asyncio.Event() + entered = 0 async def _blocking_smembers(name, *_args, **_kwargs): - events.append(("smembers_start", name)) - first_smembers.set() - await asyncio.sleep(3600) # block until cancelled - events.append(("smembers_end", name)) # unreachable once cancelled - return set() + nonlocal entered + events.append(("cmd", "smembers", name)) + entered += 1 + if entered == 3: + all_blocked.set() + try: + await asyncio.sleep(3600) + return {b"key1"} + finally: + events.append(("unwind", name)) async def _delete(*args, **_kwargs): - events.append(("delete_start", args[-1])) + events.append(("cmd", "delete", args[-1])) return 1 + async def _aexit(*_args, **_kwargs): + events.append(("aexit",)) + return False + conn.smembers = AsyncMock(side_effect=_blocking_smembers) conn.delete = AsyncMock(side_effect=_delete) - - original_aexit = conn.__aexit__ - - async def _tracking_aexit(*args, **kwargs): - events.append(("aexit", None)) - return await original_aexit(*args, **kwargs) - - conn.__aexit__ = AsyncMock(side_effect=_tracking_aexit) + conn.__aexit__ = AsyncMock(side_effect=_aexit) with _patch_redis(conn): - task = asyncio.ensure_future(layer.invalidate_by_tags(["t1", "t2"])) - await first_smembers.wait() # a child is now mid-smembers + task = asyncio.create_task(layer.invalidate_by_tags(["t1", "t2", "t3"])) + try: + await asyncio.wait_for(all_blocked.wait(), timeout=5) + except asyncio.TimeoutError: + task.cancel() + try: + await task + except BaseException: + pass + pytest.fail( + f"tags were not issued concurrently; only {entered} in flight" + ) task.cancel() with pytest.raises(asyncio.CancelledError): await task - # The connection context manager exited, and nothing issued a command - # after it did. - assert ("aexit", None) in events - aexit_index = events.index(("aexit", None)) - after_exit = events[aexit_index + 1 :] - assert not any( - kind in ("smembers_start", "delete_start") for kind, _ in after_exit - ) + kinds = [e[0] for e in events] + assert "aexit" in kinds, f"connection never closed: {events}" + aexit_idx = kinds.index("aexit") + + # Every child finished its finally path before the connection closed. + assert kinds.count("unwind") == 3, f"not all children unwound: {events}" + assert all( + i < aexit_idx for i, e in enumerate(events) if e[0] == "unwind" + ), f"a child unwound after conn close: {events}" + + # No Redis command was issued after the connection closed. + assert all( + i < aexit_idx for i, e in enumerate(events) if e[0] == "cmd" + ), f"command issued after conn close: {events}" async def test_invalidate_failure_drains_in_flight_work(self): """A failing tag returns 0 with no per-tag task left in flight.""" From 9ee059ce484b740f2526e17ae586785889d625f1 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 2 Aug 2026 18:46:53 +0000 Subject: [PATCH 4/4] fix(security): neutralize CR/LF log forging in StructuredFormatter (CWE-117) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit StructuredFormatter did not sanitize the rendered log record, so any user-controlled value carrying CR/LF — interpolated into a message, an `exc_info=True` traceback (`str(exc)`), or a structured `extra` field — could splice a forged log line (CWE-117). This was confirmed against rendered handler output, not just a helper's return value. Fix at the single formatter chokepoint: escape CR/LF and the other line separators in the fully rendered record so every sink is covered at once and each record occupies exactly one physical line. Escaping (rather than dropping) keeps the original text greppable. Adds focused regression tests asserting on rendered stream output for the message vector, the traceback vector, all targeted separators, and that a clean payload is left unescaped. Sibling to the per-sink `_safe_log()` hardening tracked on #810/#913: this addresses the traceback/extras vector centrally on `main`, independent of that PR's router-scoped change. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01Mb4JRePtEQKAgaDEZHicqA --- .../backend/config/logging_config.py | 28 +++++- tests/unit/test_logging_config_crlf.py | 94 +++++++++++++++++++ 2 files changed, 120 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..04523038f 100644 --- a/src/youtube_extension/backend/config/logging_config.py +++ b/src/youtube_extension/backend/config/logging_config.py @@ -14,6 +14,26 @@ from datetime import datetime from pathlib import Path +# CWE-117 (log injection / forging): characters that can splice or corrupt log +# lines. Every substring that reaches the final rendered record — the +# interpolated message, structured ``extra`` fields, and appended exception +# tracebacks — is neutralized at the single formatter chokepoint below, so no +# user-controlled data can forge a new log line regardless of the call site. +# We escape (rather than drop) so the original text stays greppable while each +# record is guaranteed to occupy exactly one physical line. +_LOG_FORGING_ESCAPES = { + ord("\r"): "\\r", + ord("\n"): "\\n", + ord("\v"): "\\v", + ord("\f"): "\\f", + ord("\x1c"): "\\x1c", + ord("\x1d"): "\\x1d", + ord("\x1e"): "\\x1e", + ord("\x85"): "\\x85", + ord("\u2028"): "\\u2028", + ord("\u2029"): "\\u2029", +} + class StructuredFormatter(logging.Formatter): """ @@ -36,10 +56,14 @@ 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 (includes any exc_info traceback and extras) formatted_message = super().format(record) - return formatted_message + # CWE-117: neutralize CR/LF and other line separators in the fully + # rendered record. This covers every sink at once — inline messages, + # ``exc_info=True`` tracebacks, and structured ``extra`` fields — so a + # user-controlled value carrying "\r\n" cannot forge a fake log line. + return formatted_message.translate(_LOG_FORGING_ESCAPES) 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..4db108ee9 --- /dev/null +++ b/tests/unit/test_logging_config_crlf.py @@ -0,0 +1,94 @@ +"""Regression tests for CWE-117 log-injection hardening in StructuredFormatter. + +These assert against the *rendered* handler output (not a helper's return +value), because the log-forging vector lives in the fully formatted record: +the interpolated message, ``exc_info=True`` tracebacks, and structured +``extra`` fields all reach the log stream through ``StructuredFormatter.format``. +""" + +import io +import logging + +import pytest + +from youtube_extension.backend.config.logging_config import ( + _LOG_FORGING_ESCAPES, + StructuredFormatter, +) + + +def _render(logger_name, emit): + """Render one or more log records through StructuredFormatter and return + the raw stream contents. ``emit`` receives the configured logger.""" + buf = io.StringIO() + handler = logging.StreamHandler(buf) + handler.setFormatter(StructuredFormatter("%(levelname)s - %(message)s")) + logger = logging.getLogger(logger_name) + logger.handlers[:] = [handler] + logger.propagate = False + logger.setLevel(logging.DEBUG) + emit(logger) + handler.flush() + return buf.getvalue() + + +def _payload_lines(out): + """Physical lines excluding the trailing record terminator.""" + return out.split("\n")[:-1] if out.endswith("\n") else out.split("\n") + + +@pytest.mark.unit +def test_message_crlf_cannot_forge_a_log_line(): + forged = "CRITICAL - FORGED ADMIN LINE" + out = _render( + "crlf-msg", + lambda lg: lg.info("video=%s", f"abc\r\n{forged}"), + ) + + # Exactly one record => one physical line (plus the stream terminator). + assert len(_payload_lines(out)) == 1 + # No raw CR/LF survived; they were escaped to visible sequences. + assert "\r" not in out.rstrip("\n") + assert "\\r\\n" in out + # The forged text never begins its own physical line. + assert not any(line.startswith(forged) for line in out.split("\n")) + # ...but its text is preserved (greppable) on the single record line. + assert forged in out + + +@pytest.mark.unit +def test_exc_info_traceback_cannot_forge_a_log_line(): + forged = "CRITICAL - FORGED ADMIN LINE" + + def emit(lg): + try: + raise ValueError(f"boom\r\n{forged}") + except ValueError: + lg.error("chat failed: %s", "x", exc_info=True) + + out = _render("crlf-exc", emit) + + # The whole record (message + traceback) collapses to one physical line. + assert len(_payload_lines(out)) == 1 + assert "\r" not in out.rstrip("\n") + # The exception's injected CR/LF is neutralized, so the forged text cannot + # start a new physical line even though it rode in on the traceback. + assert not any(line.startswith(forged) for line in out.split("\n")) + assert "Traceback (most recent call last):" in out + assert forged in out + + +@pytest.mark.unit +def test_all_line_separators_are_escaped(): + # Every separator the mitigation targets round-trips to a visible escape. + for cp in _LOG_FORGING_ESCAPES: + out = _render("crlf-sep", lambda lg, c=cp: lg.info("x%sy", chr(c))) + assert len(_payload_lines(out)) == 1, f"cp={cp:#x} split the record" + + +@pytest.mark.unit +def test_clean_message_is_unchanged(): + out = _render("crlf-clean", lambda lg: lg.info("ordinary message 123")) + assert "ordinary message 123" in out + # No spurious escaping of a clean payload. + assert "\\n" not in out.rstrip("\n")