diff --git a/fastapi/sse.py b/fastapi/sse.py index 1e2bd8617..229f26dae 100644 --- a/fastapi/sse.py +++ b/fastapi/sse.py @@ -203,6 +203,9 @@ def format_sse_event( The result always ends with `\n\n` (the event terminator). """ + _check_event_single_line(event) + _check_id_valid(id) + lines: list[str] = [] if comment is not None: diff --git a/tests/test_sse.py b/tests/test_sse.py index 6a9d669fe..4f92c3907 100644 --- a/tests/test_sse.py +++ b/tests/test_sse.py @@ -6,7 +6,7 @@ import fastapi.routing import pytest from fastapi import APIRouter, FastAPI from fastapi.responses import EventSourceResponse -from fastapi.sse import ServerSentEvent +from fastapi.sse import ServerSentEvent, format_sse_event from fastapi.testclient import TestClient from pydantic import BaseModel @@ -230,6 +230,35 @@ def test_server_sent_event_single_line_fields_reject_newlines( ServerSentEvent(data="test", **{field_name: value}) # ty: ignore[invalid-argument-type] +@pytest.mark.parametrize("value", ["first\nsecond", "first\rsecond", "first\r\nsecond"]) +def test_format_sse_event_rejects_multiline_event(value: str): + with pytest.raises(ValueError, match="SSE 'event' must be a single line"): + format_sse_event(data_str="safe", event=value) + + +@pytest.mark.parametrize("value", ["first\nsecond", "first\rsecond", "first\r\nsecond"]) +def test_format_sse_event_rejects_multiline_id(value: str): + with pytest.raises(ValueError, match="SSE 'id' must be a single line"): + format_sse_event(data_str="safe", id=value) + + +def test_format_sse_event_rejects_null_id(): + with pytest.raises(ValueError, match="null"): + format_sse_event(data_str="safe", id="has\0null") + + +def test_format_sse_event_validates_before_rendering_injected_event(): + with pytest.raises(ValueError, match="SSE 'event' must be a single line"): + format_sse_event(data_str="safe", event="message\ndata: injected") + + +def test_format_sse_event_accepts_valid_event_and_id(): + assert ( + format_sse_event(data_str="safe", event="message", id="event-1") + == b"event: message\ndata: safe\nid: event-1\n\n" + ) + + def test_server_sent_event_negative_retry_rejected(): with pytest.raises(ValueError): ServerSentEvent(data="test", retry=-1)