From 021bcb060e822d24a170403353126941f3577992 Mon Sep 17 00:00:00 2001
From: hemanthvnp <hemantth06@outlook.com>
Date: Sat, 30 May 2026 21:20:49 +0530
Subject: [PATCH] Fix: validate event and id fields in format_sse_event to
 prevent SSE protocol injection

PR #15588 added validation to ServerSentEvent, but the public
format_sse_event() function accepted multi-line values for event and id
without raising an error. A newline in either field splits into extra
SSE lines, allowing callers who pass user-supplied data directly to
format_sse_event() to unintentionally inject arbitrary SSE fields.

Reuse the existing _check_event_single_line and _check_id_valid helpers
at the top of format_sse_event so the two entry points enforce the same
rules. Add eight parametrised tests covering \n, \r, \r\n and \0 in
both fields, plus an end-to-end injection scenario.
---
 fastapi/sse.py    | 13 ++++++++++---
 tests/test_sse.py | 28 +++++++++++++++++++++++++++-
 2 files changed, 37 insertions(+), 4 deletions(-)

diff --git a/fastapi/sse.py b/fastapi/sse.py
index 1e2bd86171..6bca19afec 100644
--- a/fastapi/sse.py
+++ b/fastapi/sse.py
@@ -170,7 +170,7 @@ def format_sse_event(
         str | None,
         Doc(
             """
-            Optional event type name (`event:` field).
+            Optional event type name (`event:` field). Must be a single line.
             """
         ),
     ] = None,
@@ -178,7 +178,8 @@ def format_sse_event(
         str | None,
         Doc(
             """
-            Optional event ID (`id:` field).
+            Optional event ID (`id:` field). Must be a single line and must
+            not contain null (`\\0`) characters.
             """
         ),
     ] = None,
@@ -201,8 +202,14 @@ def format_sse_event(
 ) -> bytes:
     """Build SSE wire-format bytes from **pre-serialized** data.
 
-    The result always ends with `\n\n` (the event terminator).
+    Validates `event` and `id` for protocol safety (same rules as
+    `ServerSentEvent`): both must be single-line strings, and `id` must not
+    contain null characters.
+
+    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 86a67f8f9f..1c68562300 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
 
@@ -325,3 +325,29 @@ def test_no_keepalive_when_fast(client: TestClient):
     assert response.status_code == 200
     # KEEPALIVE_COMMENT is ": ping\n\n".
     assert ": ping\n" not in response.text
+
+
+# format_sse_event validation tests
+
+
+@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(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(id=value)
+
+
+def test_format_sse_event_rejects_null_id():
+    with pytest.raises(ValueError, match="null"):
+        format_sse_event(id="has\0null")
+
+
+def test_format_sse_event_injection_prevented():
+    """Newlines in event/id must not produce extra SSE field lines."""
+    with pytest.raises(ValueError):
+        format_sse_event(event="legit\ndata: injected", data_str="safe")