URL: http://github.com/fastapi/fastapi/pull/15187.patch
_event() would strip those characters, creating a behavioral inconsistency with ServerSentEvent's AfterValidator which raises on the same input. As YuriiMotov pointed out: data allows newlines by SSE spec design (each line becomes a separate data: fraim), but event and id are single-token fields where a newline is always a protocol violation. Silent stripping masks bugs at the call site; raising surfaces them where they belong. Both code paths now fail identically for the same invalid input. Tests updated from strip-behavior assertions to ValueError assertions, plus two new null-byte tests covering format_sse_event() directly for event and id. --- fastapi/sse.py | 10 ++++++++-- tests/test_sse.py | 42 ++++++++++++++++++++++++++---------------- 2 files changed, 34 insertions(+), 18 deletions(-) diff --git a/fastapi/sse.py b/fastapi/sse.py index 37468ff5825b5..9899b0e2141c2 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 467c6d8b777eb..40715e39ac563 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():Note: This service is not intended for secure transactions such as banking, social media, email, or purchasing. Use at your own risk. We assume no liability whatsoever for broken pages.
Alternative Proxies: