← pod_r5_v1i

fastapi_15588

failed WRONG_FIX wrong_fix · 16 tool calls · 147 s · fastapi/fastapi

Task input

♻️ Validate Server Sent Event fields to avoid applications from sending broken data

## Pull Request

<!--
Please start with a GitHub Discussion.

Once a team member asks you to open a PR, create it and link the discussion here.

Obvious typo fixes can be made in a PR without starting a discussion.
-->

Discussion: <!-- Link to the GitHub Discussion -->

## Description

♻️ Validate Server Sent Event fields to avoid applications from sending broken data

This was reported multiple times as a "vulnerability" by users probably using automated AI tools, with the argument that if input was taken from the request and passed directly to an event unfiltered, it could break the client.

If an app sets an `id` or `event` with multi-line content, I would consider that app broken, and probably the app itself vulnerable to something, not FastAPI. With this change, if any app sets invalid multi-line values (sourced from unfiltered input or in any other way), FastAPI will make it a server error (which it is, as the app is sending invalid data).

I don't think the code before this PR has a bug, less a vulnerability. This change is mainly a quality of life improvement to prevent badly written apps from shooting themselves in the foot by passing unsanitized data directly to a SSE.

<!-- Write the description of your PR here -->

## AI Disclaimer

<!-- If using AI, write here the prompt and model used -->

Codex with GPT-5.5

<details>
<summary>AI transcript</summary>

<!-- Paste here the entire AI transcript -->

</details>

## Checklist

- [ ] This PR is an obvious typo fix, or it links to a GitHub Discussion for the proposed code change.
- [x] I added tests for the change.
- [x] The new or updated tests fail on the main branch and pass on this PR.
- [x] Coverage stays at 100%.
- [x] The documentation explains the change if needed.

Tool calls (17)

#ToolArgumentsResult
1run_command
{
 "command": "( grep -rn 'EventSourceResponse' --include=*.py . || echo '<IMPORTANT> NO MATCH FOUND. DO NOT TRY SAME COMMAND.' ) | head -40\n( grep -rn 'sse' --include=*.py . || echo '<IMPORTANT> NO MATCH FOUND. DO NOT TRY SAME COMMAND.' ) | head -40\n( grep -rn 'Server Sent Event' --include=*.py . || echo '<IMPORTANT> NO MATCH FOUND. DO NOT TRY SAME COMMAND.' ) | head -40\n"
}
{"status": "ok", "stdout": "./tests/test_sse.py:8:from fastapi.responses import EventSourceResponse\n./tests/test_sse.py:29:@app.get(\"/items/stream\", response
full
{"status": "ok", "stdout": "./tests/test_sse.py:8:from fastapi.responses import EventSourceResponse\n./tests/test_sse.py:29:@app.get(\"/items/stream\", response_class=EventSourceResponse)\n./tests/test_sse.py:35:@app.get(\"/items/stream-sync\", response_class=EventSourceResponse)\n./tests/test_sse.py:40:@app.get(\"/items/stream-no-annotation\", response_class=EventSourceResponse)\n./tests/test_sse.py:46:@app.get(\"/items/stream-sync-no-annotation\", response_class=EventSourceResponse)\n./tests/test_sse.py:51:@app.get(\"/items/stream-dict\", response_class=EventSourceResponse)\n./tests/test_sse.py:57:@app.get(\"/items/stream-sse-event\", response_class=EventSourceResponse)\n./tests/test_sse.py:65:@app.get(\"/items/stream-mixed\", response_class=EventSourceResponse)\n./tests/test_sse.py:72:@app.get(\"/items/stream-string\", response_class=EventSourceResponse)\n./tests/test_sse.py:77:@app.post(\"/items/stream-post\", response_class=EventSourceResponse)\n./tests/test_sse.py:83:@app.get(\"/items/stream-raw\", response_class=EventSourceResponse)\n./tests/test_sse.py:93:@router.get(\"/events\", response_class=EventSourceResponse)\n./tests/test_sse.py:274:@keepalive_app.get(\"/slow-async\", response_class=EventSourceResponse)\n./tests/test_sse.py:283:@keepalive_app.get(\"/slow-sync\", response_class=EventSourceResponse)\n./fastapi/responses.py:5:from fastapi.sse import EventSourceResponse as EventSourceResponse  # noqa\n./fastapi/sse.py:20:class EventSourceResponse(StreamingResponse):\n./fastapi/sse.py:23:    Use as `response_class=EventSourceResponse` on a *path operation* that uses `yield`\n./fastapi/sse.py:46:    `response_class=EventSourceResponse`, each `ServerSentEvent` is encoded\n./fastapi/routing.py:64:    EventSourceResponse,\n./fastapi/routing.py:376:    is_sse_stream = lenient_issubclass(actual_response_class, EventSourceResponse)\n./fastapi/routing.py:856:                    # EventSourceResponse (SSE).\n./fastapi/routing.py:862:                        or lenie
2run_command
{
 "command": "cat fastapi/sse.py"
}
{"status": "ok", "stdout": "from typing import Annotated, Any\n\nfrom annotated_doc import Doc\nfrom pydantic import AfterValidator, BaseModel, Field, model_val
full
{"status": "ok", "stdout": "from typing import Annotated, Any\n\nfrom annotated_doc import Doc\nfrom pydantic import AfterValidator, BaseModel, Field, model_validator\nfrom starlette.responses import StreamingResponse\n\n# Canonical SSE event schema matching the OpenAPI 3.2 spec\n# (Section 4.14.4 \"Special Considerations for Server-Sent Events\")\n_SSE_EVENT_SCHEMA: dict[str, Any] = {\n    \"type\": \"object\",\n    \"properties\": {\n        \"data\": {\"type\": \"string\"},\n        \"event\": {\"type\": \"string\"},\n        \"id\": {\"type\": \"string\"},\n        \"retry\": {\"type\": \"integer\", \"minimum\": 0},\n    },\n}\n\n\nclass EventSourceResponse(StreamingResponse):\n    \"\"\"Streaming response with `text/event-stream` media type.\n\n    Use as `response_class=EventSourceResponse` on a *path operation* that uses `yield`\n    to enable Server Sent Events (SSE) responses.\n\n    Works with **any HTTP method** (`GET`, `POST`, etc.), which makes it compatible\n    with protocols like MCP that stream SSE over `POST`.\n\n    The actual encoding logic lives in the FastAPI routing layer. This class\n    serves mainly as a marker and sets the correct `Content-Type`.\n    \"\"\"\n\n    media_type = \"text/event-stream\"\n\n\ndef _check_id_no_null(v: str | None) -> str | None:\n    if v is not None and \"\\0\" in v:\n        raise ValueError(\"SSE 'id' must not contain null characters\")\n    return v\n\n\nclass ServerSentEvent(BaseModel):\n    \"\"\"Represents a single Server-Sent Event.\n\n    When `yield`ed from a *path operation function* that uses\n    `response_class=EventSourceResponse`, each `ServerSentEvent` is encoded\n    into the [SSE wire format](https://html.spec.whatwg.org/multipage/server-sent-events.html#parsing-an-event-stream)\n    (`text/event-stream`).\n\n    If you yield a plain object (dict, Pydantic model, etc.) instead, it is\n    automatically JSON-encoded and sent as the `data:` field.\n\n    All `data` values **including plain string
3run_command
{
 "command": "sed -n '1,110p' fastapi/sse.py"
}
{"status": "ok", "stdout": "from typing import Annotated, Any\n\nfrom annotated_doc import Doc\nfrom pydantic import AfterValidator, BaseModel, Field, model_val
full
{"status": "ok", "stdout": "from typing import Annotated, Any\n\nfrom annotated_doc import Doc\nfrom pydantic import AfterValidator, BaseModel, Field, model_validator\nfrom starlette.responses import StreamingResponse\n\n# Canonical SSE event schema matching the OpenAPI 3.2 spec\n# (Section 4.14.4 \"Special Considerations for Server-Sent Events\")\n_SSE_EVENT_SCHEMA: dict[str, Any] = {\n    \"type\": \"object\",\n    \"properties\": {\n        \"data\": {\"type\": \"string\"},\n        \"event\": {\"type\": \"string\"},\n        \"id\": {\"type\": \"string\"},\n        \"retry\": {\"type\": \"integer\", \"minimum\": 0},\n    },\n}\n\n\nclass EventSourceResponse(StreamingResponse):\n    \"\"\"Streaming response with `text/event-stream` media type.\n\n    Use as `response_class=EventSourceResponse` on a *path operation* that uses `yield`\n    to enable Server Sent Events (SSE) responses.\n\n    Works with **any HTTP method** (`GET`, `POST`, etc.), which makes it compatible\n    with protocols like MCP that stream SSE over `POST`.\n\n    The actual encoding logic lives in the FastAPI routing layer. This class\n    serves mainly as a marker and sets the correct `Content-Type`.\n    \"\"\"\n\n    media_type = \"text/event-stream\"\n\n\ndef _check_id_no_null(v: str | None) -> str | None:\n    if v is not None and \"\\0\" in v:\n        raise ValueError(\"SSE 'id' must not contain null characters\")\n    return v\n\n\nclass ServerSentEvent(BaseModel):\n    \"\"\"Represents a single Server-Sent Event.\n\n    When `yield`ed from a *path operation function* that uses\n    `response_class=EventSourceResponse`, each `ServerSentEvent` is encoded\n    into the [SSE wire format](https://html.spec.whatwg.org/multipage/server-sent-events.html#parsing-an-event-stream)\n    (`text/event-stream`).\n\n    If you yield a plain object (dict, Pydantic model, etc.) instead, it is\n    automatically JSON-encoded and sent as the `data:` field.\n\n    All `data` values **including plain string
4run_command
{
 "command": "cat fastapi/sse.py"
}
{"status": "ok", "stdout": "from typing import Annotated, Any\n\nfrom annotated_doc import Doc\nfrom pydantic import AfterValidator, BaseModel, Field, model_val
full
{"status": "ok", "stdout": "from typing import Annotated, Any\n\nfrom annotated_doc import Doc\nfrom pydantic import AfterValidator, BaseModel, Field, model_validator\nfrom starlette.responses import StreamingResponse\n\n# Canonical SSE event schema matching the OpenAPI 3.2 spec\n# (Section 4.14.4 \"Special Considerations for Server-Sent Events\")\n_SSE_EVENT_SCHEMA: dict[str, Any] = {\n    \"type\": \"object\",\n    \"properties\": {\n        \"data\": {\"type\": \"string\"},\n        \"event\": {\"type\": \"string\"},\n        \"id\": {\"type\": \"string\"},\n        \"retry\": {\"type\": \"integer\", \"minimum\": 0},\n    },\n}\n\n\nclass EventSourceResponse(StreamingResponse):\n    \"\"\"Streaming response with `text/event-stream` media type.\n\n    Use as `response_class=EventSourceResponse` on a *path operation* that uses `yield`\n    to enable Server Sent Events (SSE) responses.\n\n    Works with **any HTTP method** (`GET`, `POST`, etc.), which makes it compatible\n    with protocols like MCP that stream SSE over `POST`.\n\n    The actual encoding logic lives in the FastAPI routing layer. This class\n    serves mainly as a marker and sets the correct `Content-Type`.\n    \"\"\"\n\n    media_type = \"text/event-stream\"\n\n\ndef _check_id_no_null(v: str | None) -> str | None:\n    if v is not None and \"\\0\" in v:\n        raise ValueError(\"SSE 'id' must not contain null characters\")\n    return v\n\n\nclass ServerSentEvent(BaseModel):\n    \"\"\"Represents a single Server-Sent Event.\n\n    When `yield`ed from a *path operation function* that uses\n    `response_class=EventSourceResponse`, each `ServerSentEvent` is encoded\n    into the [SSE wire format](https://html.spec.whatwg.org/multipage/server-sent-events.html#parsing-an-event-stream)\n    (`text/event-stream`).\n\n    If you yield a plain object (dict, Pydantic model, etc.) instead, it is\n    automatically JSON-encoded and sent as the `data:` field.\n\n    All `data` values **including plain string
5run_command
{
 "command": "sed -n '110,150p' fastapi/sse.py"
}
{"status": "ok", "stdout": " retry: Annotated[\n int | None,\n Field(ge=0),\n Doc(\n \"\"\"\n Optional reconnection time in **milliseconds**.\n\n Tells the brow
full
{"status": "ok", "stdout": "    retry: Annotated[\n        int | None,\n        Field(ge=0),\n        Doc(\n            \"\"\"\n            Optional reconnection time in **milliseconds**.\n\n            Tells the browser how long to wait before reconnecting after the\n            connection is lost. Must be a non-negative integer.\n            \"\"\"\n        ),\n    ] = None\n    comment: Annotated[\n        str | None,\n        Doc(\n            \"\"\"\n            Optional comment line(s).\n\n            Comment lines start with `:` in the SSE wire format and are ignored by\n            `EventSource` clients. Useful for keep-alive pings to prevent\n            proxy/load-balancer timeouts.\n            \"\"\"\n        ),\n    ] = None\n\n    @model_validator(mode=\"after\")\n    def _check_data_exclusive(self) -> \"ServerSentEvent\":\n        if self.data is not None and self.raw_data is not None:\n            raise ValueError(\n                \"Cannot set both 'data' and 'raw_data' on the same \"\n                \"ServerSentEvent. Use 'data' for JSON-serialized payloads \"\n                \"or 'raw_data' for pre-formatted strings.\"\n            )\n        return self\n\n\ndef format_sse_event(\n    *,\n    data_str: Annotated[\n        str | None,\n        Doc(\n", "stderr": "", "exit_code": 0}
6run_command
{
 "command": "sed -n '1,150p' fastapi/sse.py"
}
{"status": "ok", "stdout": "from typing import Annotated, Any\n\nfrom annotated_doc import Doc\nfrom pydantic import AfterValidator, BaseModel, Field, model_val
full
{"status": "ok", "stdout": "from typing import Annotated, Any\n\nfrom annotated_doc import Doc\nfrom pydantic import AfterValidator, BaseModel, Field, model_validator\nfrom starlette.responses import StreamingResponse\n\n# Canonical SSE event schema matching the OpenAPI 3.2 spec\n# (Section 4.14.4 \"Special Considerations for Server-Sent Events\")\n_SSE_EVENT_SCHEMA: dict[str, Any] = {\n    \"type\": \"object\",\n    \"properties\": {\n        \"data\": {\"type\": \"string\"},\n        \"event\": {\"type\": \"string\"},\n        \"id\": {\"type\": \"string\"},\n        \"retry\": {\"type\": \"integer\", \"minimum\": 0},\n    },\n}\n\n\nclass EventSourceResponse(StreamingResponse):\n    \"\"\"Streaming response with `text/event-stream` media type.\n\n    Use as `response_class=EventSourceResponse` on a *path operation* that uses `yield`\n    to enable Server Sent Events (SSE) responses.\n\n    Works with **any HTTP method** (`GET`, `POST`, etc.), which makes it compatible\n    with protocols like MCP that stream SSE over `POST`.\n\n    The actual encoding logic lives in the FastAPI routing layer. This class\n    serves mainly as a marker and sets the correct `Content-Type`.\n    \"\"\"\n\n    media_type = \"text/event-stream\"\n\n\ndef _check_id_no_null(v: str | None) -> str | None:\n    if v is not None and \"\\0\" in v:\n        raise ValueError(\"SSE 'id' must not contain null characters\")\n    return v\n\n\nclass ServerSentEvent(BaseModel):\n    \"\"\"Represents a single Server-Sent Event.\n\n    When `yield`ed from a *path operation function* that uses\n    `response_class=EventSourceResponse`, each `ServerSentEvent` is encoded\n    into the [SSE wire format](https://html.spec.whatwg.org/multipage/server-sent-events.html#parsing-an-event-stream)\n    (`text/event-stream`).\n\n    If you yield a plain object (dict, Pydantic model, etc.) instead, it is\n    automatically JSON-encoded and sent as the `data:` field.\n\n    All `data` values **including plain string
7run_command
{
 "command": "sed -n '150,200p' fastapi/sse.py"
}
{"status": "ok", "stdout": " Doc(\n \"\"\"\n Pre-serialized data string to use as the `data:` field.\n \"\"\"\n ),\n ] = None,\n event: Annotated[\n str | None,
full
{"status": "ok", "stdout": "        Doc(\n            \"\"\"\n            Pre-serialized data string to use as the `data:` field.\n            \"\"\"\n        ),\n    ] = None,\n    event: Annotated[\n        str | None,\n        Doc(\n            \"\"\"\n            Optional event type name (`event:` field).\n            \"\"\"\n        ),\n    ] = None,\n    id: Annotated[\n        str | None,\n        Doc(\n            \"\"\"\n            Optional event ID (`id:` field).\n            \"\"\"\n        ),\n    ] = None,\n    retry: Annotated[\n        int | None,\n        Doc(\n            \"\"\"\n            Optional reconnection time in milliseconds (`retry:` field).\n            \"\"\"\n        ),\n    ] = None,\n    comment: Annotated[\n        str | None,\n        Doc(\n            \"\"\"\n            Optional comment line(s) (`:` prefix).\n            \"\"\"\n        ),\n    ] = None,\n) -> bytes:\n    \"\"\"Build SSE wire-format bytes from **pre-serialized** data.\n\n    The result always ends with `\\n\\n` (the event terminator).\n    \"\"\"\n    lines: list[str] = []\n\n    if comment is not None:\n        for line in comment.splitlines():\n            lines.append(f\": {line}\")\n\n    if event is not None:\n        lines.append(f\"event: {event}\")\n", "stderr": "", "exit_code": 0}
8run_command
{
 "command": "sed -n '200,220p' fastapi/sse.py"
}
{"status": "ok", "stdout": " lines.append(f\"event: {event}\")\n\n if data_str is not None:\n for line in data_str.splitlines():\n lines.append(f\"data: {line}\
full
{"status": "ok", "stdout": "        lines.append(f\"event: {event}\")\n\n    if data_str is not None:\n        for line in data_str.splitlines():\n            lines.append(f\"data: {line}\")\n\n    if id is not None:\n        lines.append(f\"id: {id}\")\n\n    if retry is not None:\n        lines.append(f\"retry: {retry}\")\n\n    lines.append(\"\")\n    lines.append(\"\")\n    return \"\\n\".join(lines).encode(\"utf-8\")\n\n\n# Keep-alive comment, per the SSE spec recommendation\nKEEPALIVE_COMMENT = b\": ping\\n\\n\"\n\n# Seconds between keep-alive pings when a generator is idle.\n", "stderr": "", "exit_code": 0}
9run_command
{
 "command": "cat tests/test_sse.py"
}
{"status": "ok", "stdout": "import asyncio\nimport time\nfrom collections.abc import AsyncIterable, Iterable\n\nimport fastapi.routing\nimport pytest\nfrom fast
full
{"status": "ok", "stdout": "import asyncio\nimport time\nfrom collections.abc import AsyncIterable, Iterable\n\nimport fastapi.routing\nimport pytest\nfrom fastapi import APIRouter, FastAPI\nfrom fastapi.responses import EventSourceResponse\nfrom fastapi.sse import ServerSentEvent\nfrom fastapi.testclient import TestClient\nfrom pydantic import BaseModel\n\n\nclass Item(BaseModel):\n    name: str\n    description: str | None = None\n\n\nitems = [\n    Item(name=\"Plumbus\", description=\"A multi-purpose household device.\"),\n    Item(name=\"Portal Gun\", description=\"A portal opening device.\"),\n    Item(name=\"Meeseeks Box\", description=\"A box that summons a Meeseeks.\"),\n]\n\n\napp = FastAPI()\n\n\n@app.get(\"/items/stream\", response_class=EventSourceResponse)\nasync def sse_items() -> AsyncIterable[Item]:\n    for item in items:\n        yield item\n\n\n@app.get(\"/items/stream-sync\", response_class=EventSourceResponse)\ndef sse_items_sync() -> Iterable[Item]:\n    yield from items\n\n\n@app.get(\"/items/stream-no-annotation\", response_class=EventSourceResponse)\nasync def sse_items_no_annotation():\n    for item in items:\n        yield item\n\n\n@app.get(\"/items/stream-sync-no-annotation\", response_class=EventSourceResponse)\ndef sse_items_sync_no_annotation():\n    yield from items\n\n\n@app.get(\"/items/stream-dict\", response_class=EventSourceResponse)\nasync def sse_items_dict():\n    for item in items:\n        yield {\"name\": item.name, \"description\": item.description}\n\n\n@app.get(\"/items/stream-sse-event\", response_class=EventSourceResponse)\nasync def sse_items_event():\n    yield ServerSentEvent(data=\"hello\", event=\"greeting\", id=\"1\")\n    yield ServerSentEvent(data={\"key\": \"value\"}, event=\"json-data\", id=\"2\")\n    yield ServerSentEvent(comment=\"just a comment\")\n    yield ServerSentEvent(data=\"retry-test\", retry=5000)\n\n\n@app.get(\"/items/stream-mixed\", response_class=EventSourceResponse)\nasync def sse_items_mix
10run_command
{
 "command": "python3 -m pytest tests/test_sse.py -q"
}
{"status": "ok", "stdout": ".................. [100%]\n=============================== warnings summary ===============================\n../../../../../../../Us
full
{"status": "ok", "stdout": "..................                                                       [100%]\n=============================== warnings summary ===============================\n../../../../../../../Users/jp/repos/kaggle-gemini-coding-agent-post-training/.envs/overlays/starlette-1.6.0-py3-none-any/starlette/testclient.py:53\n  /Users/jp/repos/kaggle-gemini-coding-agent-post-training/.envs/overlays/starlette-1.6.0-py3-none-any/starlette/testclient.py:53: DeprecationWarning: The anyio.abc.BlockingPortal alias is deprecated, use anyio.from_thread.BlockingPortal instead.\n    _PortalFactoryType = Callable[[], AbstractContextManager[anyio.abc.BlockingPortal]]\n\n-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html\n18 passed, 1 warning in 1.19s\n", "stderr": "", "exit_code": 0}
11run_command
{
 "command": "python3 - <<'PY'\nfrom fastapi.sse import ServerSentEvent\nfrom pydantic import ValidationError\n\ntry:\n    ServerSentEvent(data=\"hello\", event=\"multi\\nline\")\n    print(\"FAIL: multi-line event should raise ValidationError\")\nexcept ValidationError:\n    print(\"PASS: multi-line event raised ValidationError\")\nexcept Exception as e:\n    print(f\"FAIL: unexpected exception {type(e).__name__}: {e}\")\n\ntry:\n    ServerSentEvent(data=\"hello\", id=\"multi\\nline\")\n    print(\"FAIL: multi-line id should raise ValidationError\")\nexcept ValidationError:\n    print(\"PASS: multi-line id raised ValidationError\")\nexcept Exception as e:\n    print(f\"FAIL: unexpected exception {type(e).__name__}: {e}\")\n\ntry:\n    ServerSentEvent(data=\"hello\", event=\"single\")\n    print(\"PASS: single-line event passed\")\nexcept Exception as e:\n    print(f\"FAIL: single-line event failed {type(e).__name__}: {e}\")\n\ntry:\n    ServerSentEvent(data=\"hello\", id=\"single\")\n    print(\"PASS: single-line id passed\")\nexcept Exception as e:\n    print(f\"FAIL: single-line id failed {type(e).__name__}: {e}\")\nPY\n"
}
{"status": "ok", "stdout": "FAIL: multi-line event should raise ValidationError\nFAIL: multi-line id should raise ValidationError\nPASS: single-line event passe
full
{"status": "ok", "stdout": "FAIL: multi-line event should raise ValidationError\nFAIL: multi-line id should raise ValidationError\nPASS: single-line event passed\nPASS: single-line id passed\n", "stderr": "", "exit_code": 0}
12run_command
{
 "command": "cd /workspace && python3 - <<'PY'\nfrom pathlib import Path\np = Path(\"fastapi/sse.py\")\ns = p.read_text()\n\ndef _check_no_newlines(v: str | None) -> str | None:\n    if v is not None and \"\\n\" in v:\n        raise ValueError(\"SSE field must not contain newlines\")\n    return v\n\n# We need to inject _check_no_newlines before ServerSentEvent class\n# Find the end of _check_id_no_null\n# _check_id_no_null is around line 30\n\n# Let's just replace the existing _check_id_no_null or add next to it.\n# But the task says validate \"Server Sent Event fields\" (plural), specifically event and id.\n# id already has AfterValidator(_check_id_no_null).\n# event does not.\n\n# Let's refine _check_id_no_null to _check_no_newlines and use it for both.\n# Or keep _check_id_no_null and add another validator.\n\n# The current _check_id_no_null:\n# def _check_id_no_null(v: str | None) -> str | None:\n#     if v is not None and \"\\0\" in v:\n#         raise ValueError(\"SSE 'id' must not contain null characters\")\n#     return v\n\n# Proposed:\n# def _check_no_newlines(v: str | None) -> str | None:\n#     if v is not None and \"\\n\" in v:\n#         raise ValueError(\"SSE field must not contain newlines\")\n#     return v\n\n# And for id, we keep the null check.\n\n# Actually, let's just add a general _check_no_newlines and use it as an AfterValidator for both event and id.\n# Note that id already has one. Pydantic allows multiple AfterValidators.\n\nold_check_id = \"\"\"def _check_id_no_null(v: str | None) -> str | None:\n    if v is not None and \"\\\\0\" in v:\n        raise ValueError(\"SSE 'id' must not contain null characters\")\n    return v\"\"\"\n\nnew_checks = \"\"\"def _check_id_no_null(v: str | None) -> str | None:\n    if v is not None and \"\\\\0\" in v:\n        raise ValueError(\"SSE 'id' must not contain null characters\")\n    return v\n\n\ndef _check_no_newlines(v: str | None) -> str | None:\n    if v is not None and \"\\\\n\" in v:\n        raise ValueError(\"SSE field must not contain newlines\")\n    return v\"\"\"\n\n# Replace the old function with both\ns = s.replace(old_check_id, new_checks)\n\n# Now update ServerSentEvent.event\n# event: Annotated[\n#     str | None,\n#     Doc(\n#         \"\"\"\n#         Optional event type name.\n# ...\n#     ),\n# ] = None\n\n# We need to add AfterValidator(_check_no_newlines) to event\n# and to id.\n\n# event:\n# s = s.replace('event: Annotated[', 'event: Annotated[') # no change\n\n# Let's be precise with the replacement.\n\n# For event:\n# old_event = '    event: Annotated[\\n        str | None,\\n        Doc(\\n'\n# new_event = '    event: Annotated[\\n        str | None,\\n        AfterValidator(_check_no_newlines),\\n        Doc(\\n'\n\n# For id:\n# old_id = '    id: Annotated[\\n        str | None,\\n        AfterValidator(_check_id_no_null),\\n        Doc(\\n'\n# new_id = '    id: Annotated[\\n        str | None,\\n        AfterValidator(_check_id_no_null),\\n        AfterValidator(_check_no_newlines),\\n        Doc(\\n'\n\n# Let's use a more robust way to replace these.\n\nimport re\n\n# Inject _check_no_newlines\ns = s.replace(old_check_id, new_checks)\n\n# Update event\n# Find 'event: Annotated[' and the first 'Doc(' after it.\n# The structure is:\n#     event: Annotated[\n#         str | None,\n#         Doc(\n\n# Using simple replace for event:\ns = s.replace(\n    '    event: Annotated[\\n        str | None,\\n        Doc(',\n    '    event: Annotated[\\n        str | None,\\n        AfterValidator(_check_no_newlines),\\n        Doc('\n)\n\n# Using simple replace for id:\ns = s.replace(\n    '    id: Annotated[\\n        str | None,\\n        AfterValidator(_check_id_no_null),\\n        Doc(',\n    '    id: Annotated[\\n        str | None,\\n        AfterValidator(_check_id_no_null),\\n        AfterValidator(_check_no_newlines),\\n        Doc('\n)\n\np.write_text(s)\nPY\n"
}
{"status": "ok", "stdout": "", "stderr": "", "exit_code": 0}
full
{"status": "ok", "stdout": "", "stderr": "", "exit_code": 0}
13run_command
{
 "command": "python3 - <<'PY'\nfrom fastapi.sse import ServerSentEvent\nfrom pydantic import ValidationError\n\ndef test_val(data, event=None, id=None):\n    try:\n        ServerSentEvent(data=data, event=event, id=id)\n        return \"PASS\"\n    except ValidationError as e:\n        return f\"FAIL: {e}\"\n    except Exception as e:\n        return f\"ERROR: {type(e).__name__}: {e}\"\n\nprint(f\"Single line event: {test_val('hello', event='event1')}\")\nprint(f\"Multi line event: {test_val('hello', event='event\\n1')}\")\nprint(f\"Single line id: {test_val('hello', id='id1')}\")\nprint(f\"Multi line id: {test_val('hello', id='id\\n1')}\")\nprint(f\"Null char id: {test_val('hello', id='id\\01')}\")\nPY\n"
}
{"status": "ok", "stdout": "Single line event: PASS\nMulti line event: FAIL: 1 validation error for ServerSentEvent\nevent\n Value error, SSE field must not con
full
{"status": "ok", "stdout": "Single line event: PASS\nMulti line event: FAIL: 1 validation error for ServerSentEvent\nevent\n  Value error, SSE field must not contain newlines [type=value_error, input_value='event\\n1', input_type=str]\n    For further information visit https://errors.pydantic.dev/2.13/v/value_error\nSingle line id: PASS\nMulti line id: FAIL: 1 validation error for ServerSentEvent\nid\n  Value error, SSE field must not contain newlines [type=value_error, input_value='id\\n1', input_type=str]\n    For further information visit https://errors.pydantic.dev/2.13/v/value_error\nNull char id: PASS\n", "stderr": "", "exit_code": 0}
14run_command
{
 "command": "python3 - <<'PY'\nfrom fastapi.sse import ServerSentEvent\nfrom pydantic import ValidationError\n\ndef test_val(data, event=None, id=None):\n    try:\n        ServerSentEvent(data=data, event=event, id=id)\n        return \"PASS\"\n    except ValidationError as e:\n        # We expect ValidationError for invalid input\n        return \"PASS (ValidationError)\"\n    except Exception as e:\n        return f\"ERROR: {type(e).__name__}: {e}\"\n\nprint(f\"Multi line event: {test_val('hello', event='event\\n1')}\")\nprint(f\"Multi line id: {test_val('hello', id='id\\n1')}\")\n\n# Now test for Null char id: it should still fail\ndef test_null_id():\n    try:\n        ServerSentEvent(data=\"hello\", id=\"id\\01\")\n        return \"FAIL: null char id passed\"\n    except ValidationError:\n        return \"PASS (ValidationError)\"\n    except Exception as e:\n        return f\"ERROR: {type(e).__name__}: {e}\"\n\nprint(f\"Null char id: {test_null_id()}\")\nPY\n"
}
{"status": "ok", "stdout": "Multi line event: PASS (ValidationError)\nMulti line id: PASS (ValidationError)\nNull char id: FAIL: null char id passed\n", "stderr
full
{"status": "ok", "stdout": "Multi line event: PASS (ValidationError)\nMulti line id: PASS (ValidationError)\nNull char id: FAIL: null char id passed\n", "stderr": "", "exit_code": 0}
15run_command
{
 "command": "python3 - <<'PY'\nfrom fastapi.sse import ServerSentEvent\nfrom pydantic import ValidationError\n\ndef test_null_id():\n    try:\n        ServerSentEvent(data=\"hello\", id=\"id\\x00\")\n        return \"FAIL: null char id passed\"\n    except ValidationError:\n        return \"PASS (ValidationError)\"\n    except Exception as e:\n        return f\"ERROR: {type(e).__name__}: {e}\"\n\nprint(f\"Null char id: {test_null_id()}\")\nPY\n"
}
{"status": "ok", "stdout": "Null char id: PASS (ValidationError)\n", "stderr": "", "exit_code": 0}
full
{"status": "ok", "stdout": "Null char id: PASS (ValidationError)\n", "stderr": "", "exit_code": 0}
16run_command
{
 "command": "python3 -m pytest tests/test_sse.py -q"
}
{"status": "ok", "stdout": ".................. [100%]\n=============================== warnings summary ===============================\n../../../../../../../Us
full
{"status": "ok", "stdout": "..................                                                       [100%]\n=============================== warnings summary ===============================\n../../../../../../../Users/jp/repos/kaggle-gemini-coding-agent-post-training/.envs/overlays/starlette-1.6.0-py3-none-any/starlette/testclient.py:53\n  /Users/jp/repos/kaggle-gemini-coding-agent-post-training/.envs/overlays/starlette-1.6.0-py3-none-any/starlette/testclient.py:53: DeprecationWarning: The anyio.abc.BlockingPortal alias is deprecated, use anyio.from_thread.BlockingPortal instead.\n    _PortalFactoryType = Callable[[], AbstractContextManager[anyio.abc.BlockingPortal]]\n\n-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html\n18 passed, 1 warning in 1.09s\n", "stderr": "", "exit_code": 0}
17submit_patch
{}
{"status": "ok", "patch_size": 1346, "files_changed": 1}
full
{"status": "ok", "patch_size": 1346, "files_changed": 1}

Patch

diff --git a/fastapi/sse.py b/fastapi/sse.py
index 901d82496..808becb93 100644
--- a/fastapi/sse.py
+++ b/fastapi/sse.py
@@ -39,6 +39,18 @@ def _check_id_no_null(v: str | None) -> str | None:
     return v
 
 
+def _check_no_newlines(v: str | None) -> str | None:
+    if v is not None and "\n" in v:
+        raise ValueError("SSE field must not contain newlines")
+    return v
+
+
+def _check_no_newlines(v: str | None) -> str | None:
+    if v is not None and "\n" in v:
+        raise ValueError("SSE field must not contain newlines")
+    return v
+
+
 class ServerSentEvent(BaseModel):
     """Represents a single Server-Sent Event.
 
@@ -86,6 +98,7 @@ class ServerSentEvent(BaseModel):
     ] = None
     event: Annotated[
         str | None,
+        AfterValidator(_check_no_newlines),
         Doc(
             """
             Optional event type name.
@@ -98,6 +111,7 @@ class ServerSentEvent(BaseModel):
     id: Annotated[
         str | None,
         AfterValidator(_check_id_no_null),
+        AfterValidator(_check_no_newlines),
         Doc(
             """
             Optional event ID.
@@ -155,6 +169,7 @@ def format_sse_event(
     ] = None,
     event: Annotated[
         str | None,
+        AfterValidator(_check_no_newlines),
         Doc(
             """
             Optional event type name (`event:` field).

Test output

show
..........F
=================================== FAILURES ===================================
_ test_server_sent_event_single_line_fields_reject_newlines[first\nsecond-event] _

field_name = 'event', value = 'first\nsecond'

    @pytest.mark.parametrize("field_name", ["event", "id"])
    @pytest.mark.parametrize("value", ["first\nsecond", "first\rsecond", "first\r\nsecond"])
    def test_server_sent_event_single_line_fields_reject_newlines(
        field_name: str, value: str
    ):
>       with pytest.raises(ValueError, match=f"SSE '{field_name}' must be a single line"):
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E       AssertionError: Regex pattern did not match.
E         Expected regex: "SSE 'event' must be a single line"
E         Actual message: "1 validation error for ServerSentEvent\nevent\n  Value error, SSE field must not contain newlines [type=value_error, input_value='first\\nsecond', input_type=str]\n    For further information visit https://errors.pydantic.dev/2.13/v/value_error"

tests/test_sse.py:229: AssertionError
=============================== warnings summary ===============================
../../../../../../../Users/jp/repos/kaggle-gemini-coding-agent-post-training/.envs/overlays/starlette-1.6.0-py3-none-any/starlette/testclient.py:53
  /Users/jp/repos/kaggle-gemini-coding-agent-post-training/.envs/overlays/starlette-1.6.0-py3-none-any/starlette/testclient.py:53: DeprecationWarning: The anyio.abc.BlockingPortal alias is deprecated, use anyio.from_thread.BlockingPortal instead.
    _PortalFactoryType = Callable[[], AbstractContextManager[anyio.abc.BlockingPortal]]

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
!!!!!!!!!!!!!!!!!!!!!!!!!! stopping after 1 failures !!!!!!!!!!!!!!!!!!!!!!!!!!!
1 failed, 10 passed, 1 warning in 0.69s