fix(event-publisher): sanitise untrusted values before logging

CHANGELOG.md

@@ -16,6 +16,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - `bandit` now emits SARIF and uploads it to code scanning, so Python SAST findings are tracked as alerts (with dedup and dismissal history) instead of only failing the CI check. The check still fails the build on any finding.
 - Dependabot now applies a conservative `cooldown` to version updates: 30 days for major, 14 for minor and 7 for patch releases (14 days for GitHub Actions, which supports only `default-days`). New releases soak before adoption, giving yanked releases and supply-chain issues time to surface. Security updates are unaffected and still open pull requests immediately.
 
+### Security
+
+- Untrusted values are sanitised before being written to the Event Publisher log: control characters (including CR and LF) are stripped and long values truncated. This closes CodeQL alert `py/log-injection` (CWE-117) and, more importantly, hardens the ASN.1 encoding-error path, where field names and values from an otherwise unvalidated request body reached the log and could have been used to forge log entries.
+
 ## [2.2.0] - 2026-06-22
 
 ### Added

deployment/eventPublisher/EventPublisher.py

@@ -41,6 +41,25 @@
 # Reject oversized request bodies (default 1 MiB).
 MAX_CONTENT_LENGTH = getattr(config, "MAX_CONTENT_LENGTH", 1 * 1024 * 1024)
 
+# --- Log sanitisation ---
+# C0 controls, DEL and the C1 range. Removing these stops a caller from forging or
+# corrupting log entries with data that reaches a log sink (CWE-117).
+_LOG_CONTROL_RE = re.compile(r"[\x00-\x1f\x7f-\x9f]")
+# Bound a single logged value so a large payload cannot flood the log.
+LOG_VALUE_MAX_LENGTH = getattr(config, "LOG_VALUE_MAX_LENGTH", 256)
+
+
+def sanitize_for_log(value, max_length=None):
+    """Render an untrusted value safe to embed in a plain-text log line."""
+    max_length = LOG_VALUE_MAX_LENGTH if max_length is None else max_length
+    text = str(value)
+    if len(text) > max_length:
+        text = text[:max_length] + "...[truncated]"
+    # The regex above already strips CR and LF. The explicit replace() calls are kept
+    # because they are the sanitisation pattern CodeQL's py/log-injection query
+    # recognises as a taint barrier -- do not "simplify" them away, or the alert reopens.
+    return _LOG_CONTROL_RE.sub("", text).replace("\r", "").replace("\n", "")
+
 
 def build_encoder(data_folder=None):
     """Compile the ASN.1 schemas used to UPER-encode messages."""
@@ -116,13 +135,15 @@ def publish_message(sub_service, sub_service_group, geohash):
         geo_level = str(len(geohash))
         geohash_path = "/".join(list(geohash))
         key = f"v2x/{sub_service}/{sub_service_group}/g{geo_level}/{geohash_path}"
-        app.logger.info("key: %s", key)
+        app.logger.info("key: %s", sanitize_for_log(key))
 
         message_type = sub_service.upper()
         try:
             encoded = encoder.encode(message_type, data)
         except asn1tools.codecs.EncodeError as exc:
-            app.logger.warning("Encoding failed: %s", exc)
+            # The exception message embeds field names and values from the request body,
+            # which is unvalidated beyond being a JSON object -- sanitise before logging.
+            app.logger.warning("Encoding failed: %s", sanitize_for_log(exc))
             return jsonify({"error": "payload does not conform to the message schema"}), 400
 
         message = hexlify(encoded).decode("ascii")

deployment/eventPublisher/tests/test_event_publisher.py

@@ -1,3 +1,4 @@
+import logging
 import sys
 from unittest.mock import MagicMock
 
@@ -97,3 +98,85 @@ def test_configure_logging_writes_logfile(tmp_path, monkeypatch):
     monkeypatch.chdir(tmp_path)
     EventPublisher.configure_logging()
     assert (tmp_path / "app.log").exists()
+
+
+# --- Log sanitisation (CWE-117) ---
+
+
+class _RecordCollector(logging.Handler):
+    """Capture records emitted by a specific logger, independent of caplog propagation."""
+
+    def __init__(self):
+        super().__init__()
+        self.records = []
+
+    def emit(self, record):
+        self.records.append(record)
+
+
+def test_sanitize_for_log_strips_newlines():
+    result = EventPublisher.sanitize_for_log("first\r\n2026-01-01 INFO forged entry")
+    assert "\r" not in result
+    assert "\n" not in result
+    # Only the line breaks go; the surrounding text is preserved for debuggability.
+    assert result == "first2026-01-01 INFO forged entry"
+
+
+def test_sanitize_for_log_strips_control_chars():
+    # NUL, ESC (ANSI escape sequences) and a C1 control.
+    result = EventPublisher.sanitize_for_log("a\x00b\x1b[31mc\x85d")
+    assert result == "ab[31mcd"
+
+
+def test_sanitize_for_log_truncates():
+    result = EventPublisher.sanitize_for_log("x" * 1000)
+    assert result == "x" * EventPublisher.LOG_VALUE_MAX_LENGTH + "...[truncated]"
+
+
+def test_sanitize_for_log_accepts_non_str():
+    assert EventPublisher.sanitize_for_log(ValueError("boom\nsecond")) == "boomsecond"
+
+
+def test_encode_error_message_is_sanitized():
+    """The ASN.1 error message embeds values from the unvalidated request body."""
+    encoder = MagicMock()
+    encoder.encode.side_effect = asn1tools.codecs.EncodeError("bad\r\n2026-01-01 INFO forged entry")
+    app = EventPublisher.create_app(producer=MagicMock(), encoder=encoder)
+
+    collector = _RecordCollector()
+    previous_level = app.logger.level
+    app.logger.addHandler(collector)
+    app.logger.setLevel(logging.WARNING)
+    try:
+        resp = app.test_client().post("/api/publish/denm/public/7y0191k4", json={"a": 1})
+    finally:
+        app.logger.removeHandler(collector)
+        app.logger.setLevel(previous_level)
+
+    assert resp.status_code == 400
+    warnings = [r for r in collector.records if r.levelno == logging.WARNING]
+    assert len(warnings) == 1
+    message = warnings[0].getMessage()
+    assert "\r" not in message
+    assert "\n" not in message
+    assert "forged entry" in message
+
+
+def test_publish_logs_sanitized_key():
+    encoder = MagicMock()
+    encoder.encode.return_value = b"\x01\x02\x03"
+    app = EventPublisher.create_app(producer=MagicMock(), encoder=encoder)
+
+    collector = _RecordCollector()
+    previous_level = app.logger.level
+    app.logger.addHandler(collector)
+    app.logger.setLevel(logging.INFO)
+    try:
+        resp = app.test_client().post("/api/publish/denm/public/7y0191k4", json={"a": 1})
+    finally:
+        app.logger.removeHandler(collector)
+        app.logger.setLevel(previous_level)
+
+    assert resp.status_code == 200
+    key_lines = [r.getMessage() for r in collector.records if r.getMessage().startswith("key: ")]
+    assert key_lines == ["key: v2x/denm/public/g8/7/y/0/1/9/1/k/4"]