From 4d891f933e9b66da9e0da21d2329cefd82f82060 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 19:06:02 +0000 Subject: [PATCH] fix(security): build JSON log records with json.dumps (CWE-117 #1429) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The #1270 hardening escapes every line/record separator but deliberately not the double-quote, because `_UNSAFE_LOG_CHARS` is applied to a fully rendered record where escaping `"` would destroy the JSON skeleton. With `enable_json_logging` on, records were printf-interpolated into a JSON template, so a `"` in attacker content closed the `message` field and opened arbitrary new ones: log.info('benign", "level": "DEBUG", "forged": "yes') emits at INFO and parses as DEBUG, with an injected `forged` field. Most parsers take the last value on a duplicate key, so an attacker can downgrade their own entries below an alerting threshold. Escaping cannot be fixed in place: at that point attacker content and the template's structural quotes are the same characters. The fix is ordering. `StructuredFormatter` gains `json_output`; when set, it builds a dict and serializes with `json.dumps`, so escaping happens per value before any structural quote exists. `ensure_ascii=True` covers every separator the table did, including NEL/LS/PS, so records stay one physical line. The line-oriented path is untouched and keeps using the escape table. The module comment no longer claims that table makes JSON safe. Non-vacuous: removing only `"json_output": enable_json_logging` from the dictConfig fails test_setup_logging_wires_json_output_to_the_formatter and nothing else — that wiring is what a future edit could silently drop. Verified: 19 passed in tests/unit/test_logging_config_crlf.py (9 pre-existing unchanged); ruff clean on the changed files, with the repo's 2 pre-existing findings unchanged. Closes #1429 Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01MsrR4ngeBsCT9qiBftEWbB --- .../backend/config/logging_config.py | 98 ++++++++++- tests/unit/test_logging_config_crlf.py | 161 ++++++++++++++++++ 2 files changed, 250 insertions(+), 9 deletions(-) diff --git a/src/youtube_extension/backend/config/logging_config.py b/src/youtube_extension/backend/config/logging_config.py index 3d061dd92..957a52224 100644 --- a/src/youtube_extension/backend/config/logging_config.py +++ b/src/youtube_extension/backend/config/logging_config.py @@ -7,12 +7,14 @@ Provides consistent log formatting, multiple handlers, and performance monitoring. """ +import json import logging import logging.config import os import sys from datetime import datetime from pathlib import Path +from typing import Any # 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`` @@ -22,10 +24,21 @@ # # 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. +# control sequences). Each is escaped to a ``\uXXXX`` sequence — not a Python +# ``\v``/``\x1b`` shorthand — so the neutralized record remains a single +# physical line for line-oriented sinks. +# +# SCOPE: this table applies to the **line-oriented** formats only. It operates +# on a fully-rendered record, where attacker content and the template's own +# structural characters are already indistinguishable, so it can neutralize +# separators but cannot defend JSON structure. It deliberately does NOT escape +# ``"``: doing so here would corrupt the JSON skeleton rather than protect it. +# +# JSON records do not use this path at all. They are built field-by-field and +# serialized with ``json.dumps`` (see ``StructuredFormatter._format_json``), +# which escapes quotes, backslashes and every separator above *within values*, +# so structure cannot be forged. See #1429 for the field-forgery bug that came +# from applying this table to rendered JSON. # # Backslash is escaped FIRST (see ``sanitize_log_record``) so the encoding is # unambiguous and reversible: a real newline becomes a backslash-u-000a escape, @@ -50,10 +63,13 @@ 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. + Escapes CR/LF (and every other line separator, plus ESC) to ``\\uXXXX`` + sequences so attacker-controlled content cannot forge, corrupt, or split + downstream log lines (CWE-117). Backslash is escaped first, so the + transform is unambiguous and reversible. + + This is for **line-oriented** records. It does not, and cannot, make a + rendered JSON record safe — see the module comment and ``#1429``. """ return rendered.translate(_UNSAFE_LOG_CHARS) @@ -61,8 +77,18 @@ def sanitize_log_record(rendered: str) -> str: class StructuredFormatter(logging.Formatter): """ Custom formatter for structured logging with enhanced metadata. + + When ``json_output`` is set, records are assembled as a dict and + serialized with ``json.dumps`` instead of being interpolated into a JSON + template. That ordering is the security property: escaping happens per + *value*, before the structural quotes exist, so a ``"`` in a message + cannot terminate a field, add one, or shadow an earlier one. """ + def __init__(self, *args: Any, json_output: bool = False, **kwargs: Any) -> None: + super().__init__(*args, **kwargs) + self.json_output = json_output + def format(self, record: logging.LogRecord) -> str: """Format log record with structured data""" @@ -79,6 +105,9 @@ def format(self, record: logging.LogRecord) -> str: if hasattr(record, 'request_id'): record.correlation_id = record.request_id + if self.json_output: + return self._format_json(record) + # Format the base message formatted_message = super().format(record) @@ -88,6 +117,48 @@ def format(self, record: logging.LogRecord) -> str: # even when inline sanitization was not applied at the call site. return sanitize_log_record(formatted_message) + def _format_json(self, record: logging.LogRecord) -> str: + """Build the record as a dict and serialize it with ``json.dumps``. + + CWE-117: every attacker-reachable value (`message`, the `exception` + traceback, `stack_info`) enters as a dict value, so ``json.dumps`` + escapes it as string content. A ``"`` becomes ``\\"`` inside the value + and cannot reach the structural layer. + + ``ensure_ascii=True`` (the default, stated here because the guarantee + depends on it) escapes every separator ``_UNSAFE_LOG_CHARS`` covers: + the C0 controls as ``\\n``/``\\r``/``\\uXXXX``, and NEL, LS and PS as + non-ASCII ``\\uXXXX``. The record therefore stays a single physical + line, which is the same guarantee the line-oriented path provides. + """ + payload: dict[str, Any] = { + "timestamp": self.formatTime(record, self.datefmt), + "service": record.service_name, + "version": record.version, + "level": record.levelname, + "logger": record.name, + "message": record.getMessage(), + "module": record.filename, + "line": record.lineno, + "function": record.funcName, + "process": record.process, + } + + if record.exc_info: + payload["exception"] = self.formatException(record.exc_info) + if record.stack_info: + payload["stack_info"] = self.formatStack(record.stack_info) + + # Optional enrichments, only when the call site supplied them. + for attribute in ("performance_ms", "correlation_id"): + if hasattr(record, attribute): + payload[attribute] = getattr(record, attribute) + + # `default=str` keeps a non-serializable `extra` value from raising + # inside the logging path, where an exception would be swallowed and + # the record lost entirely. + return json.dumps(payload, ensure_ascii=True, default=str) + def formatException(self, ei) -> str: """Format exception with enhanced stack trace""" result = super().formatException(ei) @@ -125,6 +196,11 @@ def setup_logging( simple_format = "%(asctime)s - %(levelname)s - %(message)s" + # Retained as the documented field schema, NOT as the rendering path. + # `StructuredFormatter._format_json` builds these fields as a dict and + # serializes them; interpolating attacker content into this template is + # exactly the field-forgery bug fixed in #1429. Keep the two in step when + # adding a field. json_format = ( '{"timestamp": "%(asctime)s", "service": "%(service_name)s", ' '"version": "%(version)s", "level": "%(levelname)s", ' @@ -147,7 +223,11 @@ def setup_logging( "structured": { "()": StructuredFormatter, "format": log_format, - "datefmt": "%Y-%m-%d %H:%M:%S" + "datefmt": "%Y-%m-%d %H:%M:%S", + # Selects dict-then-`json.dumps` assembly over interpolation + # into `json_format`. Without this the JSON template is filled + # by printf and a `"` in a message forges fields (#1429). + "json_output": enable_json_logging }, "simple": { "format": simple_format, diff --git a/tests/unit/test_logging_config_crlf.py b/tests/unit/test_logging_config_crlf.py index 0dc17d72c..0d63496ac 100644 --- a/tests/unit/test_logging_config_crlf.py +++ b/tests/unit/test_logging_config_crlf.py @@ -23,6 +23,7 @@ _UNSAFE_LOG_CHARS, StructuredFormatter, sanitize_log_record, + setup_logging, ) pytestmark = [pytest.mark.unit, pytest.mark.security] @@ -165,3 +166,163 @@ def test_sanitize_log_record_escapes_each_separator_to_json_unicode(): # 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 + + +# --------------------------------------------------------------------------- +# CWE-117 field forgery in JSON records (#1429) +# +# The tests above cover separators. None of them types a `"`, which is why +# they all passed against the vulnerable code: `_UNSAFE_LOG_CHARS` +# deliberately omits the double-quote, so interpolating a message into the +# JSON *template* let attacker content close a field and open new ones. +# `test_json_logging_output_stays_parseable` came closest -- it already +# asserts `parsed["level"] == "INFO"` -- and would have caught this had its +# payload contained a quote. +# +# The fix is ordering: build a dict, then `json.dumps`, so escaping happens +# per value before any structural quote exists. +# --------------------------------------------------------------------------- + +# The exact payload from #1429: no newline, no backslash, only a quote. +_FORGERY = 'benign", "level": "DEBUG", "forged": "yes' + + +def _make_json_logger(name: str) -> tuple[logging.Logger, io.StringIO]: + buf = io.StringIO() + handler = logging.StreamHandler(buf) + handler.setFormatter( + StructuredFormatter(datefmt="%Y-%m-%d %H:%M:%S", json_output=True) + ) + logger = logging.getLogger(name) + logger.handlers[:] = [handler] + logger.setLevel(logging.DEBUG) + logger.propagate = False + return logger, buf + + +def test_quote_in_message_cannot_forge_a_json_field(): + logger, buf = _make_json_logger("json-forgery") + logger.info(_FORGERY) + parsed = json.loads(buf.getvalue()) + + # Emitted at INFO; it must not parse as DEBUG. + assert parsed["level"] == "INFO" + # No field the format never defined. + assert "forged" not in parsed + # And the payload survives intact as a *value*. + assert parsed["message"] == _FORGERY + + +def test_quote_in_lazy_args_cannot_forge_a_json_field(): + # %-interpolation happens inside getMessage(), so args are an equally + # attacker-reachable path into `message`. + logger, buf = _make_json_logger("json-forgery-args") + logger.info("user=%s", _FORGERY) + parsed = json.loads(buf.getvalue()) + + assert parsed["level"] == "INFO" + assert "forged" not in parsed + assert parsed["message"] == f"user={_FORGERY}" + + +def test_quote_in_traceback_cannot_forge_a_json_field(): + logger, buf = _make_json_logger("json-forgery-exc") + try: + raise ValueError(_FORGERY) + except ValueError: + logger.error("operation failed", exc_info=True) + parsed = json.loads(buf.getvalue()) + + assert parsed["level"] == "ERROR" + assert "forged" not in parsed + # The traceback is carried as its own value, not spliced into the record. + assert _FORGERY in parsed["exception"] + + +def test_backslash_cannot_smuggle_a_quote_out_of_a_value(): + # A trailing backslash before the quote is the classic way to defeat a + # naive escaper that handles `"` but not `\`. + logger, buf = _make_json_logger("json-forgery-backslash") + logger.info('trailing\\", "level": "DEBUG') + parsed = json.loads(buf.getvalue()) + + assert parsed["level"] == "INFO" + + +def test_json_record_stays_a_single_physical_line(): + # The separator guarantee the line-oriented path provides must survive the + # move to json.dumps -- including NEL/LS/PS, which depend on ensure_ascii. + logger, buf = _make_json_logger("json-separators") + nasty = "".join(chr(c) for c in _UNSAFE_LOG_CHARS if c != ord("\\")) + logger.info(nasty) + out = buf.getvalue() + + assert out.count("\n") == 1 # only the handler's terminator + assert out.isascii() # NEL / U+2028 / U+2029 escaped, not emitted raw + assert json.loads(out)["message"] == nasty # lossless round-trip + + +def test_benign_json_record_is_valid_and_faithful(): + # Guards the regression the naive fix caused: adding `"` to the escape + # table destroyed the JSON skeleton even for harmless messages. + logger, buf = _make_json_logger("json-benign") + logger.warning("all good %s", "video-123") + parsed = json.loads(buf.getvalue()) + + assert parsed["message"] == "all good video-123" + assert parsed["level"] == "WARNING" + assert parsed["logger"] == "json-benign" + assert parsed["service"] == "youtube-extension-api" + assert isinstance(parsed["line"], int) + + +def test_json_metadata_is_authoritative_under_attack(): + # Every field a downstream consumer routes or alerts on must reflect what + # the logger emitted, not what the message claimed. + logger, buf = _make_json_logger("json-authority") + logger.critical('x", "logger": "innocent", "timestamp": "1970-01-01 00:00:00') + parsed = json.loads(buf.getvalue()) + + assert parsed["level"] == "CRITICAL" + assert parsed["logger"] == "json-authority" + assert not parsed["timestamp"].startswith("1970") + + +def test_line_oriented_path_is_untouched_by_the_json_fix(): + # json_output defaults to False, so the existing formatter contract holds. + logger, buf = _make_logger("json-default-off") + logger.info("all good %s", "video-123") + assert buf.getvalue() == "INFO - all good video-123\n" + + +@pytest.fixture +def _restore_root_logging(): + """`setup_logging` calls dictConfig, which mutates global logging state.""" + root = logging.getLogger() + saved_handlers, saved_level = root.handlers[:], root.level + yield + root.handlers[:] = saved_handlers + root.setLevel(saved_level) + + +@pytest.mark.parametrize("enable_json", [True, False]) +def test_setup_logging_wires_json_output_to_the_formatter( + tmp_path, _restore_root_logging, enable_json +): + # The formatter is only safe on the JSON path if `setup_logging` actually + # selects it. Dropping `"json_output": enable_json_logging` from the + # dictConfig would silently restore #1429 while every formatter-level test + # above kept passing, so pin the wiring itself. + setup_logging( + log_level="INFO", + log_file=str(tmp_path / "wiring.log"), + enable_json_logging=enable_json, + ) + + formatters = [ + handler.formatter + for handler in logging.getLogger().handlers + if isinstance(handler.formatter, StructuredFormatter) + ] + assert formatters, "expected StructuredFormatter on the root logger" + assert all(f.json_output is enable_json for f in formatters)