From b393c8fc4df7690540bc7b2decce6019337cf52d Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 2 Aug 2026 19:30:03 +0000 Subject: [PATCH] fix(security): neutralize CR/LF in rendered log records (CWE-117) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit StructuredFormatter only prepended an "Exception Details:" header and never stripped line separators, so any exc_info=True / logger.exception(...) sink — and structured `extra` fields the format references — could forge, corrupt, or split downstream log lines with attacker-controlled separators, even where the message itself was sanitized inline. This central fix escapes every str.splitlines() boundary (LF, CR, VT, FF, FS, GS, RS, NEL, LS, PS) plus ESC in the FINAL rendered record, covering every sink at once without touching call sites. Escapes are emitted as JSON-valid \uXXXX sequences so JSON logging (enable_json_logging=True) stays parseable, and backslash is escaped first so the transform is unambiguous and reversible. Adds tests/unit/test_logging_config_crlf.py asserting against rendered handler output (message, exc_info traceback, logger.exception, extra fields), JSON-log parseability, reversibility, and full splitlines-boundary coverage. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_01AFJ9na5UBaskoD4eAe1fhq --- .../backend/config/logging_config.py | 49 ++++- tests/unit/test_logging_config_crlf.py | 167 ++++++++++++++++++ 2 files changed, 215 insertions(+), 1 deletion(-) 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..3d061dd92 100644 --- a/src/youtube_extension/backend/config/logging_config.py +++ b/src/youtube_extension/backend/config/logging_config.py @@ -14,6 +14,49 @@ from datetime import datetime from pathlib import Path +# Characters that can be abused to forge or corrupt log records (CWE-117 log +# injection). Any of these in dynamic content — a log message, an ``exc_info`` +# traceback, ``str(exc)``, or a structured ``extra`` field — could otherwise +# inject what looks like an independent log line, or (for JSON logs) break the +# record so downstream parsers drop or corrupt it. +# +# The set is the union of every separator ``str.splitlines()`` recognizes as a +# line boundary (LF, CR, VT, FF, FS, GS, RS, NEL, LS, PS) plus ESC (terminal +# control sequences). Each is escaped to a JSON-valid ``\uXXXX`` sequence — not +# a Python ``\v``/``\x1b`` shorthand — so the neutralized record stays valid +# JSON when ``enable_json_logging`` is on, while remaining a single physical +# line for line-oriented sinks. +# +# Backslash is escaped FIRST (see ``sanitize_log_record``) so the encoding is +# unambiguous and reversible: a real newline becomes a backslash-u-000a escape, +# while a literal backslash in the source is doubled, so the two never collide +# and the original text can be recovered by reversing the table. +_UNSAFE_LOG_CHARS = { + ord("\\"): "\\\\", + ord("\n"): "\\u000a", + ord("\r"): "\\u000d", + ord("\v"): "\\u000b", # VT / 0x0B + ord("\f"): "\\u000c", # FF / 0x0C + ord("\x1b"): "\\u001b", # ESC — terminal control / escape sequences + ord("\x1c"): "\\u001c", # FS — file separator + ord("\x1d"): "\\u001d", # GS — group separator + ord("\x1e"): "\\u001e", # RS — record separator + 0x85: "\\u0085", # NEL — Unicode next line + 0x2028: "\\u2028", # LINE SEPARATOR + 0x2029: "\\u2029", # PARAGRAPH SEPARATOR +} + + +def sanitize_log_record(rendered: str) -> str: + """Neutralize line/record separators in a fully-rendered log record. + + Escapes CR/LF (and every other line separator, plus ESC) to JSON-valid + ``\\uXXXX`` sequences so attacker-controlled content cannot forge, corrupt, + or split downstream log lines — including JSON logs (CWE-117). Backslash is + escaped first, so the transform is unambiguous and reversible. + """ + return rendered.translate(_UNSAFE_LOG_CHARS) + class StructuredFormatter(logging.Formatter): """ @@ -39,7 +82,11 @@ def format(self, record: logging.LogRecord) -> str: # Format the base message formatted_message = super().format(record) - return formatted_message + # CWE-117: neutralize line/record separators in the FINAL rendered + # record so message text, exc_info tracebacks, and any structured + # `extra` fields cannot forge, corrupt, or split downstream log lines + # even when inline sanitization was not applied at the call site. + return sanitize_log_record(formatted_message) 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..0dc17d72c --- /dev/null +++ b/tests/unit/test_logging_config_crlf.py @@ -0,0 +1,167 @@ +"""CWE-117 regression tests for StructuredFormatter log-injection hardening. + +These assert against the *rendered* handler output (not the return value of an +inline sanitizer), because the vulnerability lived in the paths that inline +sanitization does not cover: `logger.error(..., exc_info=True)`, +`logger.exception(...)`, and structured `extra` fields whose text is appended +to the record by the formatter/framework rather than the message string. +""" + +from __future__ import annotations + +import io +import json +import logging +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "src")) + +from youtube_extension.backend.config.logging_config import ( # noqa: E402 + _UNSAFE_LOG_CHARS, + StructuredFormatter, + sanitize_log_record, +) + +pytestmark = [pytest.mark.unit, pytest.mark.security] + +# A format that mirrors a realistic record: a level/message prefix an attacker +# would try to forge a second, fake copy of. +_FMT = "%(levelname)s - %(message)s" + +# The JSON format used by the module in production (`enable_json_logging=True`). +_JSON_FMT = '{"level": "%(levelname)s", "message": "%(message)s", "logger": "%(name)s"}' + + +def _make_logger(name: str, fmt: str = _FMT) -> tuple[logging.Logger, io.StringIO]: + buf = io.StringIO() + handler = logging.StreamHandler(buf) + handler.setFormatter(StructuredFormatter(fmt)) + logger = logging.getLogger(name) + logger.handlers[:] = [handler] + logger.setLevel(logging.DEBUG) + logger.propagate = False + return logger, buf + + +def _forged_line_present(rendered: str) -> bool: + """A forged line is any line *after* the first that looks like its own record.""" + tail = rendered.split("\n", 1)[1] if "\n" in rendered else "" + return "FORGED ADMIN LINE" in tail + + +def test_message_crlf_cannot_forge_a_new_line(): + logger, buf = _make_logger("crlf-message") + logger.info("user said: %s", "hello\r\nCRITICAL - FORGED ADMIN LINE") + out = buf.getvalue() + + assert "\r" not in out + # Exactly one physical line (plus the trailing newline the handler adds). + assert out.count("\n") == 1 + assert not _forged_line_present(out) + # The payload is preserved, just neutralized to JSON-valid escapes. + assert "\\u000d\\u000aCRITICAL - FORGED ADMIN LINE" in out + + +def test_exc_info_traceback_cannot_forge_log_lines(): + logger, buf = _make_logger("crlf-excinfo") + try: + raise ValueError("boom\r\nCRITICAL - FORGED ADMIN LINE") + except ValueError: + logger.error("Error in chat endpoint", exc_info=True) + out = buf.getvalue() + + assert "\r" not in out + # The entire record — message + multi-line traceback — is one physical line. + assert out.count("\n") == 1 + assert not _forged_line_present(out) + # Traceback content is retained in escaped form (still debuggable). + assert "Exception Details:" in out + assert "ValueError" in out + + +def test_logger_exception_helper_is_also_covered(): + logger, buf = _make_logger("crlf-exception-helper") + try: + raise RuntimeError("nope\nERROR - FORGED ADMIN LINE") + except RuntimeError: + logger.exception("handler failed") + out = buf.getvalue() + + assert out.endswith("\n") # only the handler's trailing newline + assert out.count("\n") == 1 # single physical record — no injected line + assert not _forged_line_present(out) + + +def test_extra_fields_referenced_by_format_are_sanitized(): + logger, buf = _make_logger( + "crlf-extra", "%(levelname)s - %(message)s - url=%(video_url)s" + ) + logger.error( + "bad request", + extra={"video_url": "http://x/\r\nCRITICAL - FORGED ADMIN LINE"}, + ) + out = buf.getvalue() + + assert "\r" not in out + assert out.count("\n") == 1 + assert not _forged_line_present(out) + + +def test_all_splitlines_boundaries_are_escaped(): + # Every character str.splitlines() treats as a line boundary must be + # neutralized, or a downstream reader that uses splitlines() could still be + # tricked into seeing multiple records. + payload = "".join(chr(c) for c in _UNSAFE_LOG_CHARS if c != ord("\\")) + assert len(payload.splitlines()) > 1 # sanity: these really are boundaries + cleaned = sanitize_log_record(payload) + assert cleaned.splitlines() == [cleaned] # collapses to a single line + + +def test_json_logging_output_stays_parseable(): + # With enable_json_logging=True the record is interpolated into a JSON + # string; the escapes must be JSON-valid so an attacker cannot corrupt or + # drop downstream JSON logs. + logger, buf = _make_logger("crlf-json", _JSON_FMT) + nasty = "line1\r\nline2\x1b[31m\x0bvt\x1crs" + logger.info("%s", nasty) + out = buf.getvalue().strip() + + parsed = json.loads(out) # must not raise + assert parsed["level"] == "INFO" + # Round-trips back to the original text: the escapes are lossless. + assert parsed["message"] == nasty + + +def test_encoding_is_reversible(): + # A real separator and a literal backslash-escape of it must not collide. + real_newline = "a\nb" + literal_text = "a\\nb" # the two characters backslash + n + assert sanitize_log_record(real_newline) != sanitize_log_record(literal_text) + # Backslash is doubled, so the mapping can be inverted unambiguously. + assert sanitize_log_record(literal_text) == "a\\\\nb" + assert sanitize_log_record(real_newline) == "a\\u000ab" + + +def test_benign_records_are_unchanged(): + logger, buf = _make_logger("crlf-benign") + logger.info("all good %s", "video-123") + out = buf.getvalue() + assert out == "INFO - all good video-123\n" + + +def test_sanitize_log_record_escapes_each_separator_to_json_unicode(): + # Build the input from the table itself so no raw separator is typed by hand + # (which is easy to get wrong for U+2028 / U+2029). + specials = {c: repl for c, repl in _UNSAFE_LOG_CHARS.items() if c != ord("\\")} + raw = "".join(chr(c) for c in specials) + cleaned = sanitize_log_record(raw) + + assert cleaned == "".join(specials[c] for c in specials) + for c in specials: + assert chr(c) not in cleaned # nothing raw remains + # The escaped blob is a JSON-valid string body that decodes losslessly back + # to the original characters (raw has no backslash, so no ambiguity). + assert json.loads(f'"{cleaned}"') == raw