Sanitize ANSI escape sequences in CSP report logging

mesop/server/static_file_serving.py

@@ -44,6 +44,30 @@ def noop():
   pass
 
 
+# Matches ANSI CSI, OSC, and other common ANSI/VT100 escape sequences so they
+# can be stripped from untrusted data before it's written to the terminal.
+_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: object) -> str:
+  # Prevent ANSI/VT100 terminal escape injection when logging
+  # attacker-controlled data (e.g. CSP report fields).
+  if not isinstance(value, str):
+    value = str(value)
+  return _ANSI_ESCAPE_RE.sub("", value)
+
+
 def configure_static_file_serving(
   app: Flask,
   static_file_runfiles_base: str,
@@ -185,14 +209,16 @@ def csp_report():
     # but it's actually application/csp-report
     report = request.get_json(force=True)
 
-    document_uri: str = report["csp-report"]["document-uri"]
+    document_uri: str = _sanitize_terminal(report["csp-report"]["document-uri"])
     path = urlparse(document_uri).path
-    blocked_uri: str = report["csp-report"]["blocked-uri"]
+    blocked_uri: str = _sanitize_terminal(report["csp-report"]["blocked-uri"])
     # 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"]
+    violated_directive: str = _sanitize_terminal(
+      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,12 @@
 
 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,
+  configure_static_file_serving,
+  gzip_cache,
+  send_file_compressed,
+)
 
 
 # Putting this test first because it's making sure the cache is empty.
@@ -73,6 +78,60 @@ def test_send_file_compressed_cached_request():
       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_removes_osc_sequences():
+  # OSC 8 hyperlink escape sequence terminated with BEL.
+  payload = "\x1b]8;;https://evil.example\x07click me\x1b]8;;\x07"
+
+  assert _sanitize_terminal(payload) == "click me"
+
+
+def test_sanitize_terminal_keeps_plain_text():
+  payload = "Normal CSP report message"
+
+  assert _sanitize_terminal(payload) == payload
+
+
+def test_sanitize_terminal_coerces_non_str_input():
+  assert _sanitize_terminal(123) == "123"
+
+
+def test_csp_report_sanitizes_ansi_escapes_in_output(capsys):
+  app = Flask(__name__)
+  configure_static_file_serving(
+    app,
+    static_file_runfiles_base="unused",
+    disable_gzip_cache=True,
+  )
+  client = app.test_client()
+
+  response = client.post(
+    "/__csp__",
+    json={
+      "csp-report": {
+        # Attacker-controlled escape sequences distinct from the app's own
+        # tc.* color codes, so this test can't pass by accident.
+        "document-uri": "https://example.com/\x1b[2J\x1b[Hpwned/page",
+        "blocked-uri": "https://evil.example\x1b]0;INJECTED-TITLE\x07",
+        "violated-directive": "connect-src",
+      }
+    },
+  )
+
+  # A NameError or other exception in the handler would surface as a 500.
+  assert response.status_code == 204
+  output = capsys.readouterr().out
+  assert "\x1b[2J" not in output
+  assert "\x1b]0;INJECTED-TITLE" not in output
+  assert "pwned/page" in output
+  assert "evil.example" in output
+
+
 if __name__ == "__main__":
   import pytest