From ca889eb161de741ff9fe591f2aba6650f63e9ccb Mon Sep 17 00:00:00 2001 From: Hayden Garvey <154503486+groupthinking@users.noreply.github.com> Date: Tue, 4 Aug 2026 03:03:35 +0000 Subject: [PATCH] fix(security): neutralize CR/LF in rendered log records (CWE-117) StructuredFormatter.format() now escapes every str.splitlines() boundary (LF, CR, VT, FF, FS, GS, RS, NEL, LS, PS) plus ESC in the final rendered record via a shared sanitize_log_record() table. Escapes are JSON-valid \uXXXX sequences with backslash escaped first, so the transform is unambiguous and reversible and JSON logs stay parseable. Covers message text, exc_info tracebacks, logger.exception, and structured extra fields. Lands PR #1270. Generated with [Linear](https://linear.app/myxstack/issue/GRV-295/land-pr-1270-fixsecurity-neutralize-crlf-in-rendered-log-records-cwe#agent-session-930110d5) Co-authored-by: linear-code[bot] <222613912+linear-code[bot]@users.noreply.github.com> --- .../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