diff --git a/fastapi/sse.py b/fastapi/sse.py index 901d824964..7e7ac671c4 100644 --- a/fastapi/sse.py +++ b/fastapi/sse.py @@ -33,9 +33,18 @@ class EventSourceResponse(StreamingResponse): media_type = "text/event-stream" -def _check_id_no_null(v: str | None) -> str | None: - if v is not None and "\0" in v: - raise ValueError("SSE 'id' must not contain null characters") +def _check_id_no_null_or_newline(v: str | None) -> str | None: + if v is not None: + if "\0" in v: + raise ValueError("SSE 'id' must not contain null characters") + if "\n" in v or "\r" in v: + raise ValueError("SSE 'id' must not contain newline characters") + return v + + +def _check_event_no_newline(v: str | None) -> str | None: + if v is not None and ("\n" in v or "\r" in v): + raise ValueError("SSE 'event' must not contain newline characters") return v @@ -86,18 +95,21 @@ class ServerSentEvent(BaseModel): ] = None event: Annotated[ str | None, + AfterValidator(_check_event_no_newline), Doc( """ Optional event type name. Maps to `addEventListener(event, ...)` on the browser. When omitted, the browser dispatches on the generic `message` event. + + **Must not contain newline (`\\n`) or carriage return (`\\r`) characters.** """ ), ] = None id: Annotated[ str | None, - AfterValidator(_check_id_no_null), + AfterValidator(_check_id_no_null_or_newline), Doc( """ Optional event ID. @@ -197,6 +209,7 @@ def format_sse_event( lines.append(f": {line}") if event is not None: + event = event.replace("\r", "").replace("\n", "") lines.append(f"event: {event}") if data_str is not None: @@ -204,6 +217,7 @@ def format_sse_event( lines.append(f"data: {line}") if id is not None: + id = id.replace("\r", "").replace("\n", "").replace("\0", "") lines.append(f"id: {id}") if retry is not None: diff --git a/tests/test_sse.py b/tests/test_sse.py index 6dfec61838..5f1ae85e37 100644 --- a/tests/test_sse.py +++ b/tests/test_sse.py @@ -221,6 +221,58 @@ def test_server_sent_event_null_id_rejected(): ServerSentEvent(data="test", id="has\0null") +def test_server_sent_event_newline_event_rejected(): + with pytest.raises(ValueError, match="newline"): + ServerSentEvent(data="test", event="chat\npwned") + + +def test_server_sent_event_cr_event_rejected(): + with pytest.raises(ValueError, match="newline"): + ServerSentEvent(data="test", event="chat\rpwned") + + +def test_server_sent_event_newline_id_rejected(): + with pytest.raises(ValueError, match="newline"): + ServerSentEvent(data="test", id="42\npwned") + + +def test_server_sent_event_cr_id_rejected(): + with pytest.raises(ValueError, match="newline"): + ServerSentEvent(data="test", id="42\rpwned") + + +def test_format_sse_event_strips_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 + + +def test_format_sse_event_strips_cr_in_event(): + 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 + + +def test_format_sse_event_strips_newline_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 + + +def test_format_sse_event_strips_cr_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 + + def test_server_sent_event_negative_retry_rejected(): with pytest.raises(ValueError): ServerSentEvent(data="test", retry=-1)