Browse Source

fix: reject newline characters in SSE event and id fields

`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)
pull/15187/head
Subhash Dasyam 4 months ago
parent
commit
8bfab526a2
  1. 22
      fastapi/sse.py
  2. 52
      tests/test_sse.py

22
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:

52
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)

Loading…
Cancel
Save