diff --git a/fastapi/sse.py b/fastapi/sse.py index 37468ff582..9899b0e214 100644 --- a/fastapi/sse.py +++ b/fastapi/sse.py @@ -212,7 +212,10 @@ def format_sse_event( lines.append(f": {line}") if event is not None: - event = event.replace("\r", "").replace("\n", "").replace("\0", "") + if "\n" in event or "\r" in event: + raise ValueError("SSE 'event' must not contain newline characters") + if "\0" in event: + raise ValueError("SSE 'event' must not contain null characters") lines.append(f"event: {event}") if data_str is not None: @@ -220,7 +223,10 @@ def format_sse_event( lines.append(f"data: {line}") if id is not None: - id = id.replace("\r", "").replace("\n", "").replace("\0", "") + if "\n" in id or "\r" in id: + raise ValueError("SSE 'id' must not contain newline characters") + if "\0" in id: + raise ValueError("SSE 'id' must not contain null characters") lines.append(f"id: {id}") if retry is not None: diff --git a/tests/test_sse.py b/tests/test_sse.py index 467c6d8b77..40715e39ac 100644 --- a/tests/test_sse.py +++ b/tests/test_sse.py @@ -246,36 +246,46 @@ def test_server_sent_event_cr_id_rejected(): ServerSentEvent(data="test", id="42\rpwned") -def test_format_sse_event_strips_newline_in_event(): +def test_format_sse_event_raises_on_newline_in_event(): from fastapi.sse import format_sse_event - result = format_sse_event(event="chat\npwned", data_str="hello") - assert b"\npwned" not in result - assert b"event: chatpwned\n" in result + with pytest.raises(ValueError, match="newline"): + format_sse_event(event="chat\npwned", data_str="hello") + + +def test_format_sse_event_raises_on_cr_in_event(): + from fastapi.sse import format_sse_event + + with pytest.raises(ValueError, match="newline"): + format_sse_event(event="chat\rpwned", data_str="hello") + + +def test_format_sse_event_raises_on_null_in_event(): + from fastapi.sse import format_sse_event + + with pytest.raises(ValueError, match="null"): + format_sse_event(event="chat\x00pwned", data_str="hello") -def test_format_sse_event_strips_cr_in_event(): +def test_format_sse_event_raises_on_newline_in_id(): from fastapi.sse import format_sse_event - result = format_sse_event(event="chat\rpwned", data_str="hello") - assert b"\rpwned" not in result - assert b"event: chatpwned\n" in result + with pytest.raises(ValueError, match="newline"): + format_sse_event(id="42\npwned", data_str="hello") -def test_format_sse_event_strips_newline_in_id(): +def test_format_sse_event_raises_on_cr_in_id(): from fastapi.sse import format_sse_event - result = format_sse_event(id="42\npwned", data_str="hello") - assert b"\npwned" not in result - assert b"id: 42pwned\n" in result + with pytest.raises(ValueError, match="newline"): + format_sse_event(id="42\rpwned", data_str="hello") -def test_format_sse_event_strips_cr_in_id(): +def test_format_sse_event_raises_on_null_in_id(): from fastapi.sse import format_sse_event - result = format_sse_event(id="42\rpwned", data_str="hello") - assert b"\rpwned" not in result - assert b"id: 42pwned\n" in result + with pytest.raises(ValueError, match="null"): + format_sse_event(id="42\x00pwned", data_str="hello") def test_server_sent_event_negative_retry_rejected():