From 06eccbf077b76309382371fe9440083bea350d54 Mon Sep 17 00:00:00 2001 From: Subhash Dasyam Date: Sun, 22 Mar 2026 11:58:45 +0400 Subject: [PATCH] fix: add null-byte protection to SSE 'event' field for symmetry with 'id' `_check_event_no_newline` lacked a `\0` check that `_check_id_no_null_or_newline` has had since the original implementation. Similarly, `format_sse_event()` stripped `\0` from `id` but not from `event`. Changes: - Extend `_check_event_no_newline` to also reject `\0` characters - Strip `\0` in `format_sse_event()` for `event` (mirrors existing `id` behavior) - Add `test_server_sent_event_null_event_rejected` regression test `event` and `id` are now fully symmetric in both the model validator and the low-level wire formatter. --- fastapi/sse.py | 9 ++++++--- tests/test_sse.py | 5 +++++ 2 files changed, 11 insertions(+), 3 deletions(-) diff --git a/fastapi/sse.py b/fastapi/sse.py index 7e7ac671c4..37468ff582 100644 --- a/fastapi/sse.py +++ b/fastapi/sse.py @@ -43,8 +43,11 @@ def _check_id_no_null_or_newline(v: str | None) -> str | None: 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") + 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 @@ -209,7 +212,7 @@ def format_sse_event( lines.append(f": {line}") if event is not None: - event = event.replace("\r", "").replace("\n", "") + event = event.replace("\r", "").replace("\n", "").replace("\0", "") lines.append(f"event: {event}") if data_str is not None: diff --git a/tests/test_sse.py b/tests/test_sse.py index 5f1ae85e37..467c6d8b77 100644 --- a/tests/test_sse.py +++ b/tests/test_sse.py @@ -221,6 +221,11 @@ 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")