Sanitize ANSI escape sequences in CSP report logging

mesop/server/static_file_serving.py

@@ -43,6 +43,29 @@
 def noop():
   pass
 
+# Matches ANSI CSI, OSC and other common ANSI escape sequences.
+# Prevent ANSI/VT100 terminal escape injection when logging
+# attacker-controlled CSP report fields.
+_ANSI_ESCAPE_RE = re.compile(
+    r"""
+    \x1B
+    (?:
+        \[[0-?]*[ -/]*[@-~]      # CSI
+      | \][^\x07\x1B]*(?:\x07|\x1B\\)  # OSC
+      | [PX^_][^\x1B]*\x1B\\     # DCS/APC/PM/SOS
+      | [@-_]                    # 2-byte escape
+    )
+    """,
+    re.VERBOSE,
+)
+
+
+def _sanitize_terminal(value: str) -> str:
+    """Remove ANSI escape sequences before logging untrusted data."""
+    if not isinstance(value, str):
+        value = str(value)
+    return _ANSI_ESCAPE_RE.sub("", value)
+
 
 def configure_static_file_serving(
   app: Flask,
@@ -185,14 +208,21 @@ def csp_report():
     # but it's actually application/csp-report
     report = request.get_json(force=True)
 
-    document_uri: str = report["csp-report"]["document-uri"]
-    path = urlparse(document_uri).path
-    blocked_uri: str = report["csp-report"]["blocked-uri"]
+    document_uri = _sanitize_terminal(
+    report["csp-report"]["document-uri"]
+    )
+
+    blocked_uri = _sanitize_terminal(
+    report["csp-report"]["blocked-uri"]
+    )
+
+    violated_directive = _sanitize_terminal(
+    report["csp-report"]["violated-directive"]
+    )
     # Remove the path from blocked_uri, keeping only the origin.
     blocked_site = (
       urlparse(blocked_uri).scheme + "://" + urlparse(blocked_uri).netloc
     )
-    violated_directive: str = report["csp-report"]["violated-directive"]
     if violated_directive == "script-src-elem":
       keyword_arg = "allowed_script_srcs"
     elif violated_directive == "connect-src":

mesop/server/static_file_serving_test.py

@@ -4,7 +4,11 @@
 
 from flask import Flask
 
-from mesop.server.static_file_serving import gzip_cache, send_file_compressed
+from mesop.server.static_file_serving import (
+  _sanitize_terminal,
+  gzip_cache,
+  send_file_compressed,
+)
 
 
 # Putting this test first because it's making sure the cache is empty.
@@ -60,19 +64,32 @@ def test_send_file_compressed_cached_request():
     gzip_file.write(cached_bytes)
   gzip_buffer.seek(0)
   gzip_cache[tmp_file_path] = gzip_buffer.getvalue()
+
   with app.test_request_context():
     response = send_file_compressed(tmp_file_path, disable_gzip_cache=False)
 
     assert response.headers["Content-Encoding"] == "gzip"
     assert response.direct_passthrough is False
     assert int(response.headers["Content-Length"]) == len(response.get_data())
 
-    # Check that the cached bytes is returned
+    # Check that the cached bytes are returned
     with gzip.GzipFile(fileobj=BytesIO(response.get_data()), mode="rb") as f:
       ungzipped_data = f.read()
       assert ungzipped_data == cached_bytes
 
 
+def test_sanitize_terminal_removes_ansi_escape_sequences():
+  payload = "\x1b[2J\x1b[H\x1b[32mPWNED\x1b[0m"
+
+  assert _sanitize_terminal(payload) == "PWNED"
+
+
+def test_sanitize_terminal_keeps_plain_text():
+  payload = "Normal CSP report message"
+
+  assert _sanitize_terminal(payload) == payload
+
+
 if __name__ == "__main__":
   import pytest