fix(security): build JSON log records with json.dumps (CWE-117 field forgery)

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``
@@ -23,10 +25,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) and NUL (which can truncate a record inside a C-based log
-# shipper). 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.
+# shipper). 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,
@@ -52,19 +65,32 @@
 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)
 
 
 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"""
 
@@ -81,6 +107,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)
 
@@ -90,6 +119,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)
@@ -127,6 +198,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", '
@@ -149,7 +225,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,

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]
@@ -179,3 +180,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)