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/src/youtube_extension/backend/services/intelligent_cache.py b/src/youtube_extension/backend/services/intelligent_cache.py index a0281de08..a92cbe171 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,51 @@ 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() + + async def _invalidate_tag(tag: str) -> int: + # 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 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 + # 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}") + + if not keys: + return 0 - for tag in tags: - # Get all keys with this tag - keys = await conn.smembers(f"uvai:tag:{tag}") - - 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..b59b3be50 100644 --- a/tests/unit/test_intelligent_cache.py +++ b/tests/unit/test_intelligent_cache.py @@ -1593,6 +1593,239 @@ 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 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 + 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_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 = 4 + conn = _make_redis_conn() + + events: list[tuple] = [] + all_blocked = asyncio.Event() + entered = 0 + + async def _blocking_smembers(name, *_args, **_kwargs): + 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(("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) + conn.__aexit__ = AsyncMock(side_effect=_aexit) + + with _patch_redis(conn): + 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 + + 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.""" + 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""" 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")