From abb4f60d189f463d6ac3892c4ab3c59900161853 Mon Sep 17 00:00:00 2001 From: Luis Alfredo Perez Medina Date: Mon, 10 Aug 2026 23:25:01 +0200 Subject: [PATCH] fix(event-publisher): sanitise untrusted values before logging CodeQL alert py/log-injection (CWE-117) flagged the `key: %s` log call, which is built from the sub_service, sub_service_group and geohash path parameters. Those three are already allowlist-validated before the log call, so that specific line was not exploitable -- CodeQL simply does not model regex-match guards as taint barriers. Auditing the alert turned up a genuine instance the scan missed three lines below: the asn1tools EncodeError message embeds field names and values taken from the request JSON body, which is only checked for being a dict. That message can carry raw CR/LF. Verified end to end against a real ASN.1 encoder -- POSTing {"header": "x\r\n"} to the unpatched service writes a fully-formed forged CRITICAL entry to app.log that is indistinguishable from a genuine one. Add a sanitize_for_log() helper that strips C0/C1 control characters and DEL, and truncates to a bounded length so a large payload cannot flood the log. Apply it at both user-derived sinks. The trailing replace() calls are redundant with the regex, but they are the pattern CodeQL recognises as a taint barrier; a comment marks them as load-bearing. Co-Authored-By: Claude Opus 5 (1M context) Signed-off-by: Luis Alfredo Perez Medina --- CHANGELOG.md | 4 + deployment/eventPublisher/EventPublisher.py | 25 +++++- .../tests/test_event_publisher.py | 83 +++++++++++++++++++ 3 files changed, 110 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 26ae8ef..39a5a7b 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/deployment/eventPublisher/EventPublisher.py b/deployment/eventPublisher/EventPublisher.py index 7199efa..9cff2aa 100644 --- a/deployment/eventPublisher/EventPublisher.py +++ b/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") diff --git a/deployment/eventPublisher/tests/test_event_publisher.py b/deployment/eventPublisher/tests/test_event_publisher.py index 6c9441f..ad76c59 100644 --- a/deployment/eventPublisher/tests/test_event_publisher.py +++ b/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"]