pFad - Phone/Frame/Anonymizer/Declutterfier! Saves Data!


--- a PPN by Garber Painting Akron. With Image Size Reduction included!

URL: http://github.com/fastapi/fastapi/pull/15187/files

" media="all" rel="stylesheet" href="https://github.githubassets.com/assets/code-2d31826944fd3be8.css" /> fix: reject newline characters in SSE event and id fields by subhashdasyam · Pull Request #15187 · fastapi/fastapi · GitHub
Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 27 additions & 4 deletions fastapi/sse.py
Original file line number Diff line number Diff line change
Expand Up @@ -33,9 +33,21 @@ 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:
if "\0" in v:
raise ValueError("SSE 'event' must not contain null characters")
if "\n" in v or "\r" in v:
raise ValueError("SSE 'event' must not contain newline characters")
return v


Expand Down Expand Up @@ -86,18 +98,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.
Expand Down Expand Up @@ -197,13 +212,21 @@ def format_sse_event(
lines.append(f": {line}")

if event is not None:
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}")

if data_str is not None:
for line in data_str.splitlines():
lines.append(f"data: {line}")

if id is not None:
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}")

if retry is not None:
Expand Down
67 changes: 67 additions & 0 deletions tests/test_sse.py
Original file line number Diff line number Diff line change
Expand Up @@ -221,6 +221,73 @@ def test_server_sent_event_null_id_rejected():
ServerSentEvent(data="test", id="has\0null")


def test_server_sent_event_null_event_rejected():
with pytest.raises(ValueError, match="null"):
ServerSentEvent(data="test", event="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_raises_on_newline_in_event():
from fastapi.sse import format_sse_event

with pytest.raises(ValueError, match="newline"):
format_sse_event(event="chat\npwned", data_str="hello")


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


Comment thread
subhashdasyam marked this conversation as resolved.
def test_format_sse_event_raises_on_newline_in_id():
from fastapi.sse import format_sse_event

with pytest.raises(ValueError, match="newline"):
format_sse_event(id="42\npwned", data_str="hello")


def test_format_sse_event_raises_on_cr_in_id():
from fastapi.sse import format_sse_event

with pytest.raises(ValueError, match="newline"):
format_sse_event(id="42\rpwned", data_str="hello")


def test_format_sse_event_raises_on_null_in_id():
from fastapi.sse import format_sse_event

with pytest.raises(ValueError, match="null"):
format_sse_event(id="42\x00pwned", data_str="hello")


def test_server_sent_event_negative_retry_rejected():
with pytest.raises(ValueError):
ServerSentEvent(data="test", retry=-1)
Expand Down
Loading
pFad - Phonifier reborn

Pfad - The Proxy pFad © 2024 Your Company Name. All rights reserved.





Check this box to remove all script contents from the fetched content.



Check this box to remove all images from the fetched content.


Check this box to remove all CSS styles from the fetched content.


Check this box to keep images inefficiently compressed and original size.

Note: This service is not intended for secure transactions such as banking, social media, email, or purchasing. Use at your own risk. We assume no liability whatsoever for broken pages.


Alternative Proxies:

Alternative Proxy

pFad Proxy

pFad v3 Proxy

pFad v4 Proxy