← pod_r2_v1e

fastapi_15588

failed WRONG_FIX wrong_fix · 25 tool calls · 280 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 (26)

#ToolArgumentsResult
1run_command
{
 "command": "grep -rn \"EventSourceResponse\" . | head -20"
}
{"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./docs/zh-hant/docs/tutorial/server-sent-events.md:38:\u8981\u5728 FastAPI \u4e2d\u4e32\u6d41 SSE\uff0c\u8acb\u5728\u4f60\u7684\u8def\u5f91\u64cd\u4f5c\u51fd\u5f0f\uff08path operation function\uff09\u4e2d\u4f7f\u7528 `yield`\uff0c\u4e26\u8a2d\u5b9a `response_class=EventSourceResponse`\u3002\n./docs/zh-hant/docs/tutorial/server-sent-events.md:40:\u5f9e `fastapi.sse` \u532f\u5165 `EventSourceResponse`\uff1a\n./docs/ja/docs/tutorial/server-sent-events.md:38:FastAPI \u3067 SSE \u3092\u30b9\u30c8\u30ea\u30fc\u30df\u30f3\u30b0\u3059\u308b\u306b\u306f\u3001*path operation \u95a2\u6570*\u3067 `yield` \u3092\u4f7f\u3044\u3001`response_class=EventSourceResponse` \u30
2run_command
{
 "command": "grep -rn \"EventSourceResponse\" fastapi | head -20"
}
{"status": "ok", "stdout": "fastapi/responses.py:5:from fastapi.sse import EventSourceResponse as EventSourceResponse # noqa\nfastapi/.agents/skills/fastapi/ref
full
{"status": "ok", "stdout": "fastapi/responses.py:5:from fastapi.sse import EventSourceResponse as EventSourceResponse  # noqa\nfastapi/.agents/skills/fastapi/references/streaming.md:16:To stream Server-Sent Events, use `response_class=EventSourceResponse` and `yield` items from the endpoint.\nfastapi/.agents/skills/fastapi/references/streaming.md:24:from fastapi.sse import EventSourceResponse\nfastapi/.agents/skills/fastapi/references/streaming.md:35:@app.get(\"/items/stream\", response_class=EventSourceResponse)\nfastapi/.agents/skills/fastapi/references/streaming.md:47:from fastapi.sse import EventSourceResponse, ServerSentEvent\nfastapi/.agents/skills/fastapi/references/streaming.md:52:@app.get(\"/events\", response_class=EventSourceResponse)\nfastapi/.agents/skills/fastapi/SKILL.md:333:See [the streaming reference](references/streaming.md) for JSON Lines, Server-Sent Events (`EventSourceResponse`, `ServerSentEvent`), and byte streaming (`StreamingResponse`) patterns.\nfastapi/sse.py:20:class EventSourceResponse(StreamingResponse):\nfastapi/sse.py:23:    Use as `response_class=EventSourceResponse` on a *path operation* that uses `yield`\nfastapi/sse.py:46:    `response_class=EventSourceResponse`, each `ServerSentEvent` is encoded\nfastapi/routing.py:64:    EventSourceResponse,\nfastapi/routing.py:376:    is_sse_stream = lenient_issubclass(actual_response_class, EventSourceResponse)\nfastapi/routing.py:856:                    # EventSourceResponse (SSE).\nfastapi/routing.py:862:                        or lenient_issubclass(response_class, EventSourceResponse)\nfastapi/routing.py:971:            response_class, EventSourceResponse\n", "stderr": "", "exit_code": 0}
3run_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
4run_command
{
 "command": "sed -n '1,120p' 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 '121,160p' fastapi/sse.py"
}
{"status": "ok", "stdout": " ] = None\n comment: Annotated[\n str | None,\n Doc(\n \"\"\"\n Optional comment line(s).\n\n Comment lines start with `:` in the SS
full
{"status": "ok", "stdout": "    ] = 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            \"\"\"\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", "stderr": "", "exit_code": 0}
6run_command
{
 "command": "sed -n '161,200p' fastapi/sse.py"
}
{"status": "ok", "stdout": " \"\"\"\n ),\n ] = None,\n id: Annotated[\n str | None,\n Doc(\n \"\"\"\n Optional event ID (`id:` field).\n \"\"\"\n ),\n ] = None,
full
{"status": "ok", "stdout": "            \"\"\"\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}
7run_command
{
 "command": "sed -n '201,220p' fastapi/sse.py"
}
{"status": "ok", "stdout": "\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.ap
full
{"status": "ok", "stdout": "\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}
8run_command
{
 "command": "python3 -m pytest tests/test_sse.py -q -x -k sse"
}
{"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.10s\n", "stderr": "", "exit_code": 0}
9run_command
{
 "command": "python3 - <<'PY'\nfrom fastapi.sse import ServerSentEvent\nimport pytest\n\ndef test_sse_multiline_event():\n    with pytest.raises(ValueError, match=\"SSE 'event' must not contain newline characters\"):\n        ServerSentEvent(event=\"event1\\nevent2\")\n\ndef test_sse_multiline_id():\n    with pytest.raises(ValueError, match=\"SSE 'id' must not contain newline characters\"):\n        ServerSentEvent(id=\"id1\\nid2\")\n\ndef test_sse_single_line_ok():\n    event = ServerSentEvent(event=\"event1\", id=\"id1\")\n    assert event.event == \"event1\"\n    assert event.id == \"id1\"\n\nif __name__ == \"__main__\":\n    try:\n        test_sse_multiline_event()\n        print(\"test_sse_multiline_event failed: no ValueError\")\n    except ValueError as e:\n        print(f\"test_sse_multiline_event passed: {e}\")\n\n    try:\n        test_sse_multiline_id()\n        print(\"test_sse_multiline_id failed: no ValueError\")\n    except ValueError as e:\n        print(f\"test_sse_multiline_id passed: {e}\")\n\n    try:\n        test_sse_single_line_ok()\n        print(\"test_sse_single_line_ok passed\")\n    except Exception as e:\n        print(f\"test_sse_single_line_ok failed: {e}\")\nPY\n"
}
{"status": "error", "error_type": "CommandError", "error_message": "Traceback (most recent call last):\n File \"<stdin>\", line 19, in <module>\n File \"<stdin>
full
{"status": "error", "error_type": "CommandError", "error_message": "Traceback (most recent call last):\n  File \"<stdin>\", line 19, in <module>\n  File \"<stdin>\", line 5, in test_sse_multiline_event\n  File \"/Users/jp/repos/kaggle-gemini-coding-agent-post-training/.envs/fastapi/lib/python3.13/site-packages/_pytest/raises.py\", line 699, in __exit__\n    fail(f\"DID NOT RAISE {self.expected_exceptions[0].__name__}\")\n    ~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n  File \"/Users/jp/repos/kaggle-gemini-coding-agent-post-training/.envs/fastapi/lib/python3.13/site-packages/_pytest/outcomes.py\", line 162, in __call__\n    raise Failed(msg=reason, pytrace=pytrace)\nFailed: DID NOT RAISE ValueError\n", "details": {"stdout": "", "stderr": "Traceback (most recent call last):\n  File \"<stdin>\", line 19, in <module>\n  File \"<stdin>\", line 5, in test_sse_multiline_event\n  File \"/Users/jp/repos/kaggle-gemini-coding-agent-post-training/.envs/fastapi/lib/python3.13/site-packages/_pytest/raises.py\", line 699, in __exit__\n    fail(f\"DID NOT RAISE {self.expected_exceptions[0].__name__}\")\n    ~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n  File \"/Users/jp/repos/kaggle-gemini-coding-agent-post-training/.envs/fastapi/lib/python3.13/site-packages/_pytest/outcomes.py\", line 162, in __call__\n    raise Failed(msg=reason, pytrace=pytrace)\nFailed: DID NOT RAISE ValueError\n", "exit_code": 1}}
10run_command
{
 "command": "cd /workspace && python3 - <<'PY'\nfrom pathlib import Path\np = Path(\"fastapi/sse.py\")\ns = p.read_text()\n\n# Define the validator for newlines\nvalidator_code = \"\"\"def _check_no_newlines(v: str | None) -> str | None:\n    if v is not None and (\"\\\\n\" in v or \"\\\\r\" in v):\n        raise ValueError(\"SSE field must not contain newline characters\")\n    return v\n\"\"\"\n\n# We need to insert the validator before ServerSentEvent class.\n# Let's find a good place. After _check_id_no_null.\nold = \"\"\"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 = old + \"\\n\\n\" + validator_code\n\n# Wait, the validator message should probably be specific to the field.\n# Let's use a more flexible validator or just add it to the AfterValidator.\n# Actually, the PR description says \"if any app sets invalid multi-line values... FastAPI will make it a server error\".\n# Pydantic's AfterValidator is good for this.\n\n# Let's redefine the validator to take the field name as an argument? No, AfterValidator doesn't work like that.\n# We can create separate validators or just one that is generic.\n# If I use one generic validator, the error message will be \"SSE field must not contain newline characters\".\n# That's probably fine.\n\n# Let's try to be more specific.\n# We can use a function that returns a validator.\n# Or just two separate functions.\n\n# Let's use two separate functions for better error messages.\nPY\n"
}
{"status": "ok", "stdout": "", "stderr": "", "exit_code": 0}
full
{"status": "ok", "stdout": "", "stderr": "", "exit_code": 0}
11run_command
{
 "command": "cd /workspace && python3 - <<'PY'\nfrom pathlib import Path\np = Path(\"fastapi/sse.py\")\ns = p.read_text()\n\n# Define the validators\nvalidators_code = \"\"\"def _check_no_newlines_event(v: str | None) -> str | None:\n    if v is not None and (\"\\\\n\" in v or \"\\\\r\" in v):\n        raise ValueError(\"SSE 'event' must not contain newline characters\")\n    return v\n\ndef _check_no_newlines_id(v: str | None) -> str | None:\n    if v is not None and (\"\\\\n\" in v or \"\\\\r\" in v):\n        raise ValueError(\"SSE 'id' must not contain newline characters\")\n    return v\n\"\"\"\n\n# Insert validators after _check_id_no_null\nold = \"\"\"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 = old + \"\\n\\n\" + validators_code\n\ns = s.replace(old, new)\n\n# Now update ServerSentEvent fields to use these validators.\n# For 'event', we add AfterValidator(_check_no_newlines_event)\n# For 'id', we add AfterValidator(_check_no_newlines_id) to the existing AfterValidator(_check_id_no_null)\n\n# Let's find the 'event' field definition\n# event: Annotated[\n#     str | None,\n#     Doc(...),\n# ] = None\n\n# We want:\n# event: Annotated[\n#     str | None,\n#     AfterValidator(_check_no_newlines_event),\n#     Doc(...),\n# ] = None\n\n# Let's use a more robust replacement.\n# For event:\nold_event = \"\"\"    event: Annotated[\n        str | None,\n        Doc(\"\"\"\nnew_event = \"\"\"    event: Annotated[\n        str | None,\n        AfterValidator(_check_no_newlines_event),\n        Doc(\"\"\"\n\n# For id:\nold_id = \"\"\"    id: Annotated[\n        str | None,\n        AfterValidator(_check_id_no_null),\n        Doc(\"\"\"\nnew_id = \"\"\"    id: Annotated[\n        str | None,\n        AfterValidator(_check_id_no_null),\n        AfterValidator(_check_no_newlines_id),\n        Doc(\"\"\"\n\ns = s.replace(old_event, new_event)\ns = s.replace(old_id, new_id)\n\np.write_text(s)\nPY\n"
}
{"status": "ok", "stdout": "", "stderr": "", "exit_code": 0}
full
{"status": "ok", "stdout": "", "stderr": "", "exit_code": 0}
12run_command
{
 "command": "python3 - <<'PY'\nfrom fastapi.sse import ServerSentEvent\nimport pytest\n\ndef test_sse_multiline_event():\n    try:\n        ServerSentEvent(event=\"event1\\nevent2\")\n        print(\"test_sse_multiline_event failed: no ValueError\")\n    except ValueError as e:\n        print(f\"test_sse_multiline_event passed: {e}\")\n\ndef test_sse_multiline_id():\n    try:\n        ServerSentEvent(id=\"id1\\nid2\")\n        print(\"test_sse_multiline_id failed: no ValueError\")\n    except ValueError as e:\n        print(f\"test_sse_multiline_id passed: {e}\")\n\ndef test_sse_single_line_ok():\n    try:\n        event = ServerSentEvent(event=\"event1\", id=\"id1\")\n        assert event.event == \"event1\"\n        assert event.id == \"id1\"\n        print(\"test_sse_single_line_ok passed\")\n    except Exception as e:\n        print(f\"test_sse_single_line_ok failed: {e}\")\n\nif __name__ == \"__main__\":\n    test_sse_multiline_event()\n    test_sse_multiline_id()\n    test_sse_single_line_ok()\nPY\n"
}
{"status": "ok", "stdout": "test_sse_multiline_event passed: 1 validation error for ServerSentEvent\nevent\n Value error, SSE 'event' must not contain newline c
full
{"status": "ok", "stdout": "test_sse_multiline_event passed: 1 validation error for ServerSentEvent\nevent\n  Value error, SSE 'event' must not contain newline characters [type=value_error, input_value='event1\\nevent2', input_type=str]\n    For further information visit https://errors.pydantic.dev/2.13/v/value_error\ntest_sse_multiline_id passed: 1 validation error for ServerSentEvent\nid\n  Value error, SSE 'id' must not contain newline characters [type=value_error, input_value='id1\\nid2', input_type=str]\n    For further information visit https://errors.pydantic.dev/2.13/v/value_error\ntest_sse_single_line_ok passed\n", "stderr": "", "exit_code": 0}
13run_command
{
 "command": "python3 -m pytest tests/test_sse.py -q -x -k sse"
}
{"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.06s\n", "stderr": "", "exit_code": 0}
14run_command
{
 "command": "git diff"
}
{"status": "ok", "stdout": "diff --git a/fastapi/sse.py b/fastapi/sse.py\nindex 901d82496..faaeabd97 100644\n--- a/fastapi/sse.py\n+++ b/fastapi/sse.py\n@@ -39,
full
{"status": "ok", "stdout": "diff --git a/fastapi/sse.py b/fastapi/sse.py\nindex 901d82496..faaeabd97 100644\n--- a/fastapi/sse.py\n+++ b/fastapi/sse.py\n@@ -39,6 +39,17 @@ def _check_id_no_null(v: str | None) -> str | None:\n     return v\n \n \n+def _check_no_newlines_event(v: str | None) -> str | None:\n+    if v is not None and (\"\\n\" in v or \"\\r\" in v):\n+        raise ValueError(\"SSE 'event' must not contain newline characters\")\n+    return v\n+\n+def _check_no_newlines_id(v: str | None) -> str | None:\n+    if v is not None and (\"\\n\" in v or \"\\r\" in v):\n+        raise ValueError(\"SSE 'id' must not contain newline characters\")\n+    return v\n+\n+\n class ServerSentEvent(BaseModel):\n     \"\"\"Represents a single Server-Sent Event.\n \n@@ -86,6 +97,7 @@ class ServerSentEvent(BaseModel):\n     ] = None\n     event: Annotated[\n         str | None,\n+        AfterValidator(_check_no_newlines_event),\n         Doc(\n             \"\"\"\n             Optional event type name.\n@@ -98,6 +110,7 @@ class ServerSentEvent(BaseModel):\n     id: Annotated[\n         str | None,\n         AfterValidator(_check_id_no_null),\n+        AfterValidator(_check_no_newlines_id),\n         Doc(\n             \"\"\"\n             Optional event ID.\n@@ -155,6 +168,7 @@ def format_sse_event(\n     ] = None,\n     event: Annotated[\n         str | None,\n+        AfterValidator(_check_no_newlines_event),\n         Doc(\n             \"\"\"\n             Optional event type name (`event:` field).\n", "stderr": "", "exit_code": 0}
15run_command
{
 "command": "sed -n '160,180p' fastapi/sse.py"
}
{"status": "ok", "stdout": " *,\n data_str: Annotated[\n str | None,\n Doc(\n \"\"\"\n Pre-serialized data string to use as the `data:` field.\n \"\"\"\n ),\n ]
full
{"status": "ok", "stdout": "    *,\n    data_str: Annotated[\n        str | None,\n        Doc(\n            \"\"\"\n            Pre-serialized data string to use as the `data:` field.\n            \"\"\"\n        ),\n    ] = None,\n    event: Annotated[\n        str | None,\n        AfterValidator(_check_no_newlines_event),\n        Doc(\n            \"\"\"\n            Optional event type name (`event:` field).\n            \"\"\"\n        ),\n    ] = None,\n    id: Annotated[\n        str | None,\n        Doc(\n", "stderr": "", "exit_code": 0}
16run_command
{
 "command": "cd /workspace && python3 - <<'PY'\nfrom pathlib import Path\np = Path(\"fastapi/sse.py\")\ns = p.read_text()\n\n# I accidentally added AfterValidator to format_sse_event arguments.\n# format_sse_event is a function, not a Pydantic model.\n# Annotated with AfterValidator is for Pydantic validation.\n# It doesn't hurt, but it's not doing anything and it's not correct for a function signature.\n\n# Let's remove it from format_sse_event.\nold = \"\"\"    event: Annotated[\n        str | None,\n        AfterValidator(_check_no_newlines_event),\n        Doc(\"\"\"\nnew = \"\"\"    event: Annotated[\n        str | None,\n        Doc(\"\"\"\n\ns = s.replace(old, new)\np.write_text(s)\nPY\n"
}
{"status": "ok", "stdout": "", "stderr": "", "exit_code": 0}
full
{"status": "ok", "stdout": "", "stderr": "", "exit_code": 0}
17run_command
{
 "command": "git diff"
}
{"status": "ok", "stdout": "diff --git a/fastapi/sse.py b/fastapi/sse.py\nindex 901d82496..82d174360 100644\n--- a/fastapi/sse.py\n+++ b/fastapi/sse.py\n@@ -39,
full
{"status": "ok", "stdout": "diff --git a/fastapi/sse.py b/fastapi/sse.py\nindex 901d82496..82d174360 100644\n--- a/fastapi/sse.py\n+++ b/fastapi/sse.py\n@@ -39,6 +39,17 @@ def _check_id_no_null(v: str | None) -> str | None:\n     return v\n \n \n+def _check_no_newlines_event(v: str | None) -> str | None:\n+    if v is not None and (\"\\n\" in v or \"\\r\" in v):\n+        raise ValueError(\"SSE 'event' must not contain newline characters\")\n+    return v\n+\n+def _check_no_newlines_id(v: str | None) -> str | None:\n+    if v is not None and (\"\\n\" in v or \"\\r\" in v):\n+        raise ValueError(\"SSE 'id' must not contain newline characters\")\n+    return v\n+\n+\n class ServerSentEvent(BaseModel):\n     \"\"\"Represents a single Server-Sent Event.\n \n@@ -98,6 +109,7 @@ class ServerSentEvent(BaseModel):\n     id: Annotated[\n         str | None,\n         AfterValidator(_check_id_no_null),\n+        AfterValidator(_check_no_newlines_id),\n         Doc(\n             \"\"\"\n             Optional event ID.\n", "stderr": "", "exit_code": 0}
18run_command
{
 "command": "python3 - <<'PY'\nfrom fastapi.sse import ServerSentEvent\nimport pytest\n\ndef test_sse_multiline_event():\n    try:\n        ServerSentEvent(event=\"event1\\nevent2\")\n        print(\"test_sse_multiline_event failed: no ValueError\")\n    except ValueError as e:\n        print(f\"test_sse_multiline_event passed: {e}\")\n\ndef test_sse_multiline_id():\n    try:\n        ServerSentEvent(id=\"id1\\nid2\")\n        print(\"test_sse_multiline_id failed: no ValueError\")\n    except ValueError as e:\n        print(f\"test_sse_multiline_id passed: {e}\")\n\ndef test_sse_single_line_ok():\n    try:\n        event = ServerSentEvent(event=\"event1\", id=\"id1\")\n        assert event.event == \"event1\"\n        assert event.id == \"id1\"\n        print(\"test_sse_single_line_ok passed\")\n    except Exception as e:\n        print(f\"test_sse_single_line_ok failed: {e}\")\n\nif __name__ == \"__main__\":\n    test_sse_multiline_event()\n    test_sse_multiline_id()\n    test_sse_single_line_ok()\nPY\n"
}
{"status": "ok", "stdout": "test_sse_multiline_event failed: no ValueError\ntest_sse_multiline_id passed: 1 validation error for ServerSentEvent\nid\n Value err
full
{"status": "ok", "stdout": "test_sse_multiline_event failed: no ValueError\ntest_sse_multiline_id passed: 1 validation error for ServerSentEvent\nid\n  Value error, SSE 'id' must not contain newline characters [type=value_error, input_value='id1\\nid2', input_type=str]\n    For further information visit https://errors.pydantic.dev/2.13/v/value_error\ntest_sse_single_line_ok passed\n", "stderr": "", "exit_code": 0}
19run_command
{
 "command": "git diff"
}
{"status": "ok", "stdout": "diff --git a/fastapi/sse.py b/fastapi/sse.py\nindex 901d82496..82d174360 100644\n--- a/fastapi/sse.py\n+++ b/fastapi/sse.py\n@@ -39,
full
{"status": "ok", "stdout": "diff --git a/fastapi/sse.py b/fastapi/sse.py\nindex 901d82496..82d174360 100644\n--- a/fastapi/sse.py\n+++ b/fastapi/sse.py\n@@ -39,6 +39,17 @@ def _check_id_no_null(v: str | None) -> str | None:\n     return v\n \n \n+def _check_no_newlines_event(v: str | None) -> str | None:\n+    if v is not None and (\"\\n\" in v or \"\\r\" in v):\n+        raise ValueError(\"SSE 'event' must not contain newline characters\")\n+    return v\n+\n+def _check_no_newlines_id(v: str | None) -> str | None:\n+    if v is not None and (\"\\n\" in v or \"\\r\" in v):\n+        raise ValueError(\"SSE 'id' must not contain newline characters\")\n+    return v\n+\n+\n class ServerSentEvent(BaseModel):\n     \"\"\"Represents a single Server-Sent Event.\n \n@@ -98,6 +109,7 @@ class ServerSentEvent(BaseModel):\n     id: Annotated[\n         str | None,\n         AfterValidator(_check_id_no_null),\n+        AfterValidator(_check_no_newlines_id),\n         Doc(\n             \"\"\"\n             Optional event ID.\n", "stderr": "", "exit_code": 0}
20run_command
{
 "command": "cd /workspace && python3 - <<'PY'\nfrom pathlib import Path\np = Path(\"fastapi/sse.py\")\ns = p.read_text()\n\n# I missed adding AfterValidator(_check_no_newlines_event) to ServerSentEvent.event\n# Let's find it.\n\n# The previous attempt failed because I replaced old_event with new_event,\n# but maybe the old_event string didn't match exactly.\n\n# Let's look at the file content again.\n# event: Annotated[\n#     str | None,\n#     Doc(\n#         \"\"\"\n#         Optional event type name.\n#         ...\n#         \"\"\"\n#     ),\n# ] = None\n\n# Let's use a more flexible replacement.\nimport re\n\n# This regex looks for the event field in ServerSentEvent and inserts AfterValidator\n# It looks for: event: Annotated[ \\n str | None, \\n Doc(\n# And replaces it with: event: Annotated[ \\n str | None, \\n AfterValidator(_check_no_newlines_event), \\n Doc(\n\n# Actually, let's just use a simple replace on a larger block.\n\n# Let's find the line number of 'event: Annotated['\nfor i, line in enumerate(s.splitlines()):\n    if 'event: Annotated[' in line:\n        print(f\"Line {i}: {line}\")\n\nPY\n"
}
{"status": "ok", "stdout": "Line 97: event: Annotated[\nLine 167: event: Annotated[\n", "stderr": "", "exit_code": 0}
full
{"status": "ok", "stdout": "Line 97:     event: Annotated[\nLine 167:     event: Annotated[\n", "stderr": "", "exit_code": 0}
21run_command
{
 "command": "sed -n '90,110p' fastapi/sse.py"
}
{"status": "ok", "stdout": " Use this when you need to send pre-formatted text, HTML fragments,\n CSV lines, or any non-JSON payload. The string is placed direc
full
{"status": "ok", "stdout": "            Use this when you need to send pre-formatted text, HTML fragments,\n            CSV lines, or any non-JSON payload. The string is placed directly\n            into the `data:` field as-is.\n\n            Mutually exclusive with `data`.\n            \"\"\"\n        ),\n    ] = None\n    event: Annotated[\n        str | None,\n        Doc(\n            \"\"\"\n            Optional event type name.\n\n            Maps to `addEventListener(event, ...)` on the browser. When omitted,\n            the browser dispatches on the generic `message` event.\n            \"\"\"\n        ),\n    ] = None\n    id: Annotated[\n        str | None,\n", "stderr": "", "exit_code": 0}
22run_command
{
 "command": "cd /workspace && python3 - <<'PY'\nfrom pathlib import Path\np = Path(\"fastapi/sse.py\")\ns = p.read_text()\n\nold = \"\"\"    event: Annotated[\n        str | None,\n        Doc(\"\"\"\nnew = \"\"\"    event: Annotated[\n        str | None,\n        AfterValidator(_check_no_newlines_event),\n        Doc(\"\"\"\n\ns = s.replace(old, new)\np.write_text(s)\nPY\n"
}
{"status": "ok", "stdout": "", "stderr": "", "exit_code": 0}
full
{"status": "ok", "stdout": "", "stderr": "", "exit_code": 0}
23run_command
{
 "command": "python3 - <<'PY'\nfrom fastapi.sse import ServerSentEvent\nimport pytest\n\ndef test_sse_multiline_event():\n    try:\n        ServerSentEvent(event=\"event1\\nevent2\")\n        print(\"test_sse_multiline_event failed: no ValueError\")\n    except ValueError as e:\n        print(f\"test_sse_multiline_event passed: {e}\")\n\ndef test_sse_multiline_id():\n    try:\n        ServerSentEvent(id=\"id1\\nid2\")\n        print(\"test_sse_multiline_id failed: no ValueError\")\n    except ValueError as e:\n        print(f\"test_sse_multiline_id passed: {e}\")\n\ndef test_sse_single_line_ok():\n    try:\n        event = ServerSentEvent(event=\"event1\", id=\"id1\")\n        assert event.event == \"event1\"\n        assert event.id == \"id1\"\n        print(\"test_sse_single_line_ok passed\")\n    except Exception as e:\n        print(f\"test_sse_single_line_ok failed: {e}\")\n\nif __name__ == \"__main__\":\n    test_sse_multiline_event()\n    test_sse_multiline_id()\n    test_sse_single_line_ok()\nPY\n"
}
{"status": "ok", "stdout": "test_sse_multiline_event passed: 1 validation error for ServerSentEvent\nevent\n Value error, SSE 'event' must not contain newline c
full
{"status": "ok", "stdout": "test_sse_multiline_event passed: 1 validation error for ServerSentEvent\nevent\n  Value error, SSE 'event' must not contain newline characters [type=value_error, input_value='event1\\nevent2', input_type=str]\n    For further information visit https://errors.pydantic.dev/2.13/v/value_error\ntest_sse_multiline_id passed: 1 validation error for ServerSentEvent\nid\n  Value error, SSE 'id' must not contain newline characters [type=value_error, input_value='id1\\nid2', input_type=str]\n    For further information visit https://errors.pydantic.dev/2.13/v/value_error\ntest_sse_single_line_ok passed\n", "stderr": "", "exit_code": 0}
24run_command
{
 "command": "python3 -m pytest tests/test_sse.py -q -x -k sse"
}
{"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.08s\n", "stderr": "", "exit_code": 0}
25run_command
{
 "command": "submit_patch"
}
{"status": "error", "error_type": "CommandError", "error_message": "/bin/bash: submit_patch: command not found\n", "details": {"stdout": "", "stderr": "/bin/bas
full
{"status": "error", "error_type": "CommandError", "error_message": "/bin/bash: submit_patch: command not found\n", "details": {"stdout": "", "stderr": "/bin/bash: submit_patch: command not found\n", "exit_code": 127}}
26submit_patch
{}
{"status": "ok", "patch_size": 1419, "files_changed": 1}
full
{"status": "ok", "patch_size": 1419, "files_changed": 1}

Patch

diff --git a/fastapi/sse.py b/fastapi/sse.py
index 901d82496..faaeabd97 100644
--- a/fastapi/sse.py
+++ b/fastapi/sse.py
@@ -39,6 +39,17 @@ def _check_id_no_null(v: str | None) -> str | None:
     return v
 
 
+def _check_no_newlines_event(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")
+    return v
+
+def _check_no_newlines_id(v: str | None) -> str | None:
+    if v is not None and ("\n" in v or "\r" in v):
+        raise ValueError("SSE 'id' must not contain newline characters")
+    return v
+
+
 class ServerSentEvent(BaseModel):
     """Represents a single Server-Sent Event.
 
@@ -86,6 +97,7 @@ class ServerSentEvent(BaseModel):
     ] = None
     event: Annotated[
         str | None,
+        AfterValidator(_check_no_newlines_event),
         Doc(
             """
             Optional event type name.
@@ -98,6 +110,7 @@ class ServerSentEvent(BaseModel):
     id: Annotated[
         str | None,
         AfterValidator(_check_id_no_null),
+        AfterValidator(_check_no_newlines_id),
         Doc(
             """
             Optional event ID.
@@ -155,6 +168,7 @@ def format_sse_event(
     ] = None,
     event: Annotated[
         str | None,
+        AfterValidator(_check_no_newlines_event),
         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 'event' must not contain newline characters [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.45s