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

src/youtube_extension/backend/config/logging_config.py

@@ -14,6 +14,26 @@
 from datetime import datetime
 from pathlib import Path
 
+# Control / line-breaking characters that must never survive verbatim into a
+# rendered log record. Left in place, a CR/LF (or any other line separator)
+# carried by user-controlled data OR by raw exception text appended via
+# ``exc_info`` lets an attacker forge or split log lines (CWE-117 log
+# injection). We escape rather than drop them so the original content stays
+# visible and greppable while no longer able to start a new physical line.
+_UNSAFE_LOG_CHARS = {
+    0x00: "\\x00",   # NUL
+    0x0A: "\\n",     # LF
+    0x0B: "\\v",     # vertical tab
+    0x0C: "\\f",     # form feed
+    0x0D: "\\r",     # CR
+    0x1C: "\\x1c",   # file separator
+    0x1D: "\\x1d",   # group separator
+    0x1E: "\\x1e",   # record separator
+    0x85: "\\x85",   # NEL (Unicode next-line)
+    0x2028: "\\u2028",  # LINE SEPARATOR
+    0x2029: "\\u2029",  # PARAGRAPH SEPARATOR
+}
+
 
 class StructuredFormatter(logging.Formatter):
     """
@@ -36,10 +56,16 @@ 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 appends the exc_info traceback, if any)
         formatted_message = super().format(record)
 
-        return formatted_message
+        # CWE-117: neutralize CR/LF and other line separators in the FINAL
+        # rendered record. This is the single central sink that covers every
+        # logger call at once — message interpolation, structured ``extra``
+        # fields, and raw ``exc_info`` tracebacks — so a value such as
+        # "boom\r\nCRITICAL - FORGED ADMIN LINE" can no longer forge a log line
+        # even when the individual call site forgot to sanitize its inputs.
+        return formatted_message.translate(_UNSAFE_LOG_CHARS)
 
     def formatException(self, ei) -> str:
         """Format exception with enhanced stack trace"""

tests/unit/test_logging_config_crlf.py

@@ -0,0 +1,89 @@
+"""Regression tests for CWE-117 log-injection hardening in StructuredFormatter.
+
+These assert against the *rendered* handler output (not a helper's return
+value), covering the two ways CR/LF can reach a log record:
+
+1. User-controlled data interpolated into the log message.
+2. Raw exception text appended via ``exc_info`` / ``logger.exception`` — the
+   sink that inline message sanitization does not reach.
+"""
+
+import io
+import logging
+
+import pytest
+
+from youtube_extension.backend.config.logging_config import (
+    _UNSAFE_LOG_CHARS,
+    StructuredFormatter,
+)
+
+FORGED = "CRITICAL - FORGED ADMIN LINE"
+
+
+def _render(logger_name, emit):
+    """Render one record through StructuredFormatter and return the raw output."""
+    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)
+    return buf.getvalue()
+
+
+def test_crlf_in_message_cannot_forge_log_lines():
+    out = _render(
+        "crlf-message",
+        lambda lg: lg.info("video_id=%s captured", f"abc\r\nINFO - {FORGED}"),
+    )
+    assert out.count("\n") == 1  # exactly one trailing newline -> one physical line
+    assert "\r" not in out
+    assert not any(line.strip() == FORGED for line in out.split("\n"))
+
+
+def test_exc_info_traceback_cannot_forge_log_lines():
+    def emit(lg):
+        try:
+            raise ValueError(f"boom\r\n{FORGED}")
+        except ValueError:
+            lg.error("Error in chat endpoint: %s", "safe-arg", exc_info=True)
+
+    out = _render("crlf-exc-info", emit)
+
+    # The whole record — message + traceback — is a single physical line.
+    assert out.count("\n") == 1
+    assert "\r" not in out
+    # The forged payload survives as inline, escaped text, never as its own line.
+    assert not any(line.strip() == FORGED for line in out.split("\n"))
+    assert "\\r\\n" in out
+
+
+def test_logger_exception_and_structured_extra_are_scrubbed():
+    def emit(lg):
+        try:
+            raise RuntimeError(f"db down\r\n{FORGED}")
+        except RuntimeError:
+            lg.exception("failure", extra={"request_id": "id\r\nWARNING - forged"})
+
+    out = _render("crlf-extra", emit)
+    assert out.count("\n") == 1
+    assert "\r" not in out
+    assert not any(line.strip() == FORGED for line in out.split("\n"))
+
+
+@pytest.mark.parametrize("ch", sorted(_UNSAFE_LOG_CHARS))
+def test_every_unsafe_char_is_escaped(ch):
+    payload = f"before{chr(ch)}after"
+    out = _render(f"crlf-charset-{ch}", lambda lg: lg.info("v=%s", payload))
+    body = out.rstrip("\n")  # drop the handler's single line terminator
+    assert chr(ch) not in body  # raw control char never survives in the record
+    assert out.count("\n") == 1
+    assert _UNSAFE_LOG_CHARS[ch] in body  # replaced by its visible escape
+
+
+def test_ordinary_message_is_unchanged():
+    out = _render("crlf-plain", lambda lg: lg.info("plain message %s", "value"))
+    assert out == "INFO - plain message value\n"