diff --git a/mesop/server/static_file_serving.py b/mesop/server/static_file_serving.py index aac56edbc..7c520e746 100644 --- a/mesop/server/static_file_serving.py +++ b/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": diff --git a/mesop/server/static_file_serving_test.py b/mesop/server/static_file_serving_test.py index 032f60b24..8d941268e 100644 --- a/mesop/server/static_file_serving_test.py +++ b/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