diff --git a/mesop/server/static_file_serving.py b/mesop/server/static_file_serving.py index aac56edbc..0c54abede 100644 --- a/mesop/server/static_file_serving.py +++ b/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": diff --git a/mesop/server/static_file_serving_test.py b/mesop/server/static_file_serving_test.py index 032f60b24..896898a27 100644 --- a/mesop/server/static_file_serving_test.py +++ b/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,6 +64,7 @@ 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) @@ -67,12 +72,24 @@ def test_send_file_compressed_cached_request(): 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