Browse Source

fix: raise ValueError in format_sse_event instead of stripping invalid chars

Replace silent .replace() sanitization in format_sse_event() with explicit
ValueError raises for \n, \r, and \0 in the event and id fields.

Previously, format_sse_event() would strip those characters, creating a
behavioral inconsistency with ServerSentEvent's AfterValidator which raises
on the same input. As YuriiMotov pointed out: data allows newlines by SSE
spec design (each line becomes a separate data: frame), but event and id are
single-token fields where a newline is always a protocol violation. Silent
stripping masks bugs at the call site; raising surfaces them where they belong.

Both code paths now fail identically for the same invalid input. Tests updated
from strip-behavior assertions to ValueError assertions, plus two new null-byte
tests covering format_sse_event() directly for event and id.
pull/15187/head
Subhash Dasyam 4 months ago
parent
commit
9e51a7d695
  1. 10
      fastapi/sse.py
  2. 42
      tests/test_sse.py

10
fastapi/sse.py

@ -212,7 +212,10 @@ def format_sse_event(
lines.append(f": {line}") lines.append(f": {line}")
if event is not None: if event is not None:
event = event.replace("\r", "").replace("\n", "").replace("\0", "") if "\n" in event or "\r" in event:
raise ValueError("SSE 'event' must not contain newline characters")
if "\0" in event:
raise ValueError("SSE 'event' must not contain null characters")
lines.append(f"event: {event}") lines.append(f"event: {event}")
if data_str is not None: if data_str is not None:
@ -220,7 +223,10 @@ def format_sse_event(
lines.append(f"data: {line}") lines.append(f"data: {line}")
if id is not None: if id is not None:
id = id.replace("\r", "").replace("\n", "").replace("\0", "") if "\n" in id or "\r" in id:
raise ValueError("SSE 'id' must not contain newline characters")
if "\0" in id:
raise ValueError("SSE 'id' must not contain null characters")
lines.append(f"id: {id}") lines.append(f"id: {id}")
if retry is not None: if retry is not None:

42
tests/test_sse.py

@ -246,36 +246,46 @@ def test_server_sent_event_cr_id_rejected():
ServerSentEvent(data="test", id="42\rpwned") ServerSentEvent(data="test", id="42\rpwned")
def test_format_sse_event_strips_newline_in_event(): def test_format_sse_event_raises_on_newline_in_event():
from fastapi.sse import format_sse_event from fastapi.sse import format_sse_event
result = format_sse_event(event="chat\npwned", data_str="hello") with pytest.raises(ValueError, match="newline"):
assert b"\npwned" not in result format_sse_event(event="chat\npwned", data_str="hello")
assert b"event: chatpwned\n" in result
def test_format_sse_event_raises_on_cr_in_event():
from fastapi.sse import format_sse_event
with pytest.raises(ValueError, match="newline"):
format_sse_event(event="chat\rpwned", data_str="hello")
def test_format_sse_event_raises_on_null_in_event():
from fastapi.sse import format_sse_event
with pytest.raises(ValueError, match="null"):
format_sse_event(event="chat\x00pwned", data_str="hello")
def test_format_sse_event_strips_cr_in_event(): def test_format_sse_event_raises_on_newline_in_id():
from fastapi.sse import format_sse_event from fastapi.sse import format_sse_event
result = format_sse_event(event="chat\rpwned", data_str="hello") with pytest.raises(ValueError, match="newline"):
assert b"\rpwned" not in result format_sse_event(id="42\npwned", data_str="hello")
assert b"event: chatpwned\n" in result
def test_format_sse_event_strips_newline_in_id(): def test_format_sse_event_raises_on_cr_in_id():
from fastapi.sse import format_sse_event from fastapi.sse import format_sse_event
result = format_sse_event(id="42\npwned", data_str="hello") with pytest.raises(ValueError, match="newline"):
assert b"\npwned" not in result format_sse_event(id="42\rpwned", data_str="hello")
assert b"id: 42pwned\n" in result
def test_format_sse_event_strips_cr_in_id(): def test_format_sse_event_raises_on_null_in_id():
from fastapi.sse import format_sse_event from fastapi.sse import format_sse_event
result = format_sse_event(id="42\rpwned", data_str="hello") with pytest.raises(ValueError, match="null"):
assert b"\rpwned" not in result format_sse_event(id="42\x00pwned", data_str="hello")
assert b"id: 42pwned\n" in result
def test_server_sent_event_negative_retry_rejected(): def test_server_sent_event_negative_retry_rejected():

Loading…
Cancel
Save