Browse Source

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.
pull/15651/head
hemanthvnp 2 weeks ago
parent
commit
021bcb060e
  1. 13
      fastapi/sse.py
  2. 28
      tests/test_sse.py

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

28
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")

Loading…
Cancel
Save