From 8bfab526a2c015765fb708dd2c63d816a4461edb Mon Sep 17 00:00:00 2001 From: Subhash Dasyam Date: Sat, 21 Mar 2026 18:56:24 +0400 Subject: [PATCH] fix: reject newline characters in SSE event and id fields MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `format_sse_event()` interpolated the `event` and `id` fields into SSE wire-format bytes without stripping or rejecting newline characters. A `\n` or `\r\n` in either field injects an additional SSE field-line into the stream — enabling event-type spoofing and fabricated data payloads in browser EventSource clients. The `data` and `comment` fields were already protected via `splitlines()`; the omission for `event` and `id` was an inconsistency, not a design decision. The identical bug class was fixed in Hono (GHSA-p6xx-57qc-3wxr). Changes: - Add `_check_event_no_newline()` validator and apply it to `ServerSentEvent.event` via `AfterValidator` - Extend `_check_id_no_null()` to also reject `\n` and `\r` - Strip `\r`/`\n` in `format_sse_event()` for `event` and `id` as defense-in-depth (low-level function, bypasses model validators) - Add 7 regression tests covering LF/CR rejection in both the model and the low-level formatting function Severity: Medium (client-side injection; server unaffected) CWE: CWE-116 — Improper Encoding for Output in a Different Plaintext Context OWASP: A03:2021 — Injection Prior art: Hono GHSA-p6xx-57qc-3wxr (same class, same fix pattern) --- fastapi/sse.py | 22 ++++++++++++++++---- tests/test_sse.py | 52 +++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 70 insertions(+), 4 deletions(-) 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)