← pod_r5_v1h

fastapi_15588

failed WRONG_FIX wrong_fix · 22 tool calls · 163 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 (23)

#ToolArgumentsResult
1run_command
{
 "command": "grep -rn \"EventSourceResponse\" /workspace | head -20"
}
{"status": "ok", "stdout": "/workspace/tests/test_sse.py:8:from fastapi.responses import EventSourceResponse\n/workspace/tests/test_sse.py:29:@app.get(\"/items/
full
{"status": "ok", "stdout": "/workspace/tests/test_sse.py:8:from fastapi.responses import EventSourceResponse\n/workspace/tests/test_sse.py:29:@app.get(\"/items/stream\", response_class=EventSourceResponse)\n/workspace/tests/test_sse.py:35:@app.get(\"/items/stream-sync\", response_class=EventSourceResponse)\n/workspace/tests/test_sse.py:40:@app.get(\"/items/stream-no-annotation\", response_class=EventSourceResponse)\n/workspace/tests/test_sse.py:46:@app.get(\"/items/stream-sync-no-annotation\", response_class=EventSourceResponse)\n/workspace/tests/test_sse.py:51:@app.get(\"/items/stream-dict\", response_class=EventSourceResponse)\n/workspace/tests/test_sse.py:57:@app.get(\"/items/stream-sse-event\", response_class=EventSourceResponse)\n/workspace/tests/test_sse.py:65:@app.get(\"/items/stream-mixed\", response_class=EventSourceResponse)\n/workspace/tests/test_sse.py:72:@app.get(\"/items/stream-string\", response_class=EventSourceResponse)\n/workspace/tests/test_sse.py:77:@app.post(\"/items/stream-post\", response_class=EventSourceResponse)\n/workspace/tests/test_sse.py:83:@app.get(\"/items/stream-raw\", response_class=EventSourceResponse)\n/workspace/tests/test_sse.py:93:@router.get(\"/events\", response_class=EventSourceResponse)\n/workspace/tests/test_sse.py:274:@keepalive_app.get(\"/slow-async\", response_class=EventSourceResponse)\n/workspace/tests/test_sse.py:283:@keepalive_app.get(\"/slow-sync\", response_class=EventSourceResponse)\n/workspace/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/workspace/docs/zh-hant/docs/tutorial/server-sent-events.md:40:\u5f9e `fastapi.sse` \u532f\u5165 `EventSourceResponse`\uff1a\n/workspace/docs/ja/docs/tutorial/server-sent-events.md:38:FastAPI \u3067 SSE \u3092\u30b9\u30c8\u30ea\u30fc\u30
2run_command
{
 "command": "find /workspace -name \"*sse*\""
}
{"status": "ok", "stdout": "/workspace/tests/test_sse.py\n/workspace/tests/test_tutorial/test_dataclasses\n/workspace/tests/test_pydanticv2_dataclasses_uuid_str
full
{"status": "ok", "stdout": "/workspace/tests/test_sse.py\n/workspace/tests/test_tutorial/test_dataclasses\n/workspace/tests/test_pydanticv2_dataclasses_uuid_stringified_annotations.py\n/workspace/docs/zh-hant/docs/advanced/dataclasses.md\n/workspace/docs/zh-hant/docs/tutorial/dependencies/classes-as-dependencies.md\n/workspace/docs/zh-hant/docs/how-to/custom-docs-ui-assets.md\n/workspace/docs/ja/docs/advanced/dataclasses.md\n/workspace/docs/ja/docs/tutorial/dependencies/classes-as-dependencies.md\n/workspace/docs/ja/docs/how-to/custom-docs-ui-assets.md\n/workspace/docs/ru/docs/advanced/dataclasses.md\n/workspace/docs/ru/docs/tutorial/dependencies/classes-as-dependencies.md\n/workspace/docs/ru/docs/how-to/custom-docs-ui-assets.md\n/workspace/docs/pt/docs/advanced/dataclasses.md\n/workspace/docs/pt/docs/tutorial/dependencies/classes-as-dependencies.md\n/workspace/docs/pt/docs/how-to/custom-docs-ui-assets.md\n/workspace/docs/zh/docs/advanced/dataclasses.md\n/workspace/docs/zh/docs/tutorial/dependencies/classes-as-dependencies.md\n/workspace/docs/zh/docs/how-to/custom-docs-ui-assets.md\n/workspace/docs/uk/docs/advanced/dataclasses.md\n/workspace/docs/uk/docs/tutorial/dependencies/classes-as-dependencies.md\n/workspace/docs/uk/docs/how-to/custom-docs-ui-assets.md\n/workspace/docs/de/docs/advanced/dataclasses.md\n/workspace/docs/de/docs/tutorial/dependencies/classes-as-dependencies.md\n/workspace/docs/de/docs/how-to/custom-docs-ui-assets.md\n/workspace/docs/ko/docs/advanced/dataclasses.md\n/workspace/docs/ko/docs/tutorial/dependencies/classes-as-dependencies.md\n/workspace/docs/ko/docs/how-to/custom-docs-ui-assets.md\n/workspace/docs/fr/docs/advanced/dataclasses.md\n/workspace/docs/fr/docs/tutorial/dependencies/classes-as-dependencies.md\n/workspace/docs/fr/docs/how-to/custom-docs-ui-assets.md\n/workspace/docs/es/docs/advanced/dataclasses.md\n/workspace/docs/es/docs/tutorial/dependencies/classes-as-dependencies.md\n/workspace/docs/es/docs/how-to/custom-docs-ui-assets.md\n
3run_command
{
 "command": "cat /workspace/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' /workspace/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' /workspace/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": "grep -n \"def _check_id_no_null\" /workspace/fastapi/sse.py"
}
{"status": "ok", "stdout": "36:def _check_id_no_null(v: str | None) -> str | None:\n", "stderr": "", "exit_code": 0}
full
{"status": "ok", "stdout": "36:def _check_id_no_null(v: str | None) -> str | None:\n", "stderr": "", "exit_code": 0}
7run_command
{
 "command": "sed -n '36,40p' /workspace/fastapi/sse.py"
}
{"status": "ok", "stdout": "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 cont
full
{"status": "ok", "stdout": "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", "stderr": "", "exit_code": 0}
8run_command
{
 "command": "cat /workspace/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
9run_command
{
 "command": "python3 -m pytest tests/test_sse.py -q -x -k sse_items_event"
}
{"status": "error", "error_type": "CommandError", "error_message": "\n=============================== warnings summary ===============================\n../../..
full
{"status": "error", "error_type": "CommandError", "error_message": "\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 deselected, 1 warning in 0.45s\n", "details": {"stdout": "\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 deselected, 1 warning in 0.45s\n", "stderr": "", "exit_code": 5}}
10run_command
{
 "command": "python3 -m pytest tests/test_sse.py -q -k \"test_async_generator_with_model\""
}
{"status": "ok", "stdout": ". [100%]\n=============================== warnings summary ===============================\n../../../../../../../Users/jp/repos/kagg
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\n1 passed, 17 deselected, 1 warning in 0.50s\n", "stderr": "", "exit_code": 0}
11run_command
{
 "command": "cd /workspace && python3 - <<'PY'\nfrom fastapi.sse import ServerSentEvent\nfrom pydantic import ValidationError\n\ntry:\n    ServerSentEvent(id=\"line1\\nline2\")\n    print(\"Failed: id with newline should raise ValidationError\")\nexcept ValidationError:\n    print(\"Passed: id with newline raised ValidationError\")\n\ntry:\n    ServerSentEvent(event=\"line1\\nline2\")\n    print(\"Failed: event with newline should raise ValidationError\")\nexcept ValidationError:\n    print(\"Passed: event with newline raised ValidationError\")\n\ntry:\n    ServerSentEvent(data=\"line1\\nline2\")\n    print(\"Passed: data with newline is allowed (it's serialized to JSON or passed as raw_data)\")\nexcept ValidationError:\n    print(\"Failed: data with newline should NOT raise ValidationError\")\nPY\n"
}
{"status": "ok", "stdout": "Failed: id with newline should raise ValidationError\nFailed: event with newline should raise ValidationError\nPassed: data with new
full
{"status": "ok", "stdout": "Failed: id with newline should raise ValidationError\nFailed: event with newline should raise ValidationError\nPassed: data with newline is allowed (it's serialized to JSON or passed as raw_data)\n", "stderr": "", "exit_code": 0}
12run_command
{
 "command": "cd /workspace && python3 - <<'PY'\nfrom fastapi.sse import ServerSentEvent\nfrom pydantic import ValidationError\n\ntry:\n    ServerSentEvent(id=\"line1\\rline2\")\n    print(\"Failed: id with CRLF should raise ValidationError\")\nexcept ValidationError:\n    print(\"Passed: id with CRLF raised ValidationError\")\n\ntry:\n    ServerSentEvent(event=\"line1\\rline2\")\n    print(\"Failed: event with CRLF should raise ValidationError\")\nexcept ValidationError:\n    print(\"Passed: event with CRLF raised ValidationError\")\nPY\n"
}
{"status": "ok", "stdout": "Failed: id with CRLF should raise ValidationError\nFailed: event with CRLF should raise ValidationError\n", "stderr": "", "exit_code
full
{"status": "ok", "stdout": "Failed: id with CRLF should raise ValidationError\nFailed: event with CRLF should raise ValidationError\n", "stderr": "", "exit_code": 0}
13run_command
{
 "command": "cd /workspace && python3 - <<'PY'\nfrom fastapi.sse import ServerSentEvent\nfrom pydantic import ValidationError\n\ntry:\n    ServerSentEvent(id=\"line1\\r\\nline2\")\n    print(\"Failed: id with CRLF should raise ValidationError\")\nexcept ValidationError:\n    print(\"Passed: id with CRLF raised ValidationError\")\nPY\n"
}
{"status": "ok", "stdout": "Failed: id with CRLF should raise ValidationError\n", "stderr": "", "exit_code": 0}
full
{"status": "ok", "stdout": "Failed: id with CRLF should raise ValidationError\n", "stderr": "", "exit_code": 0}
14run_command
{
 "command": "cd /workspace && python3 - <<'PY'\nfrom pathlib import Path\np = Path(\"fastapi/sse.py\")\ns = p.read_text()\n\n# Add a validator for no newlines\n# We can create a general validator function\n\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\"\"\"\n\n# Let's define a more general validator that check for newlines\n# and null characters.\n\nnew = \"\"\"def _check_sse_field_no_newlines(v: str | None, field_name: str) -> str | None:\n    if v is not None:\n        if \"\\\\n\" in v or \"\\\\r\" in v:\n            raise ValueError(f\"SSE '{field_name}' must not contain newline characters\")\n    return v\n\ndef _check_id_no_null(v: str | None) -> str | None:\n    if v is not None:\n        if \"\\\\0\" in v:\n            raise ValueError(\"SSE 'id' must not contain null characters\")\n        if \"\\\\n\" in v or \"\\\\r\" in v:\n            raise ValueError(\"SSE 'id' must not contain newline characters\")\n    return v\n\"\"\"\n# Actually, the problem says \"id\" or \"event\" with multi-line content.\n# Let's just add a validator that checks for \\n and \\r.\n\n# Let's redefine _check_id_no_null and add _check_event_no_newlines.\n# Or better, a helper.\n\n# Wait, let's look at the existing code again.\n# id: Annotated[str | None, AfterValidator(_check_id_no_null), ...]\n# event: Annotated[str | None, ...]\n\n# I will replace _check_id_no_null and add _check_field_no_newlines.\nPY\n"
}
{"status": "ok", "stdout": "", "stderr": "", "exit_code": 0}
full
{"status": "ok", "stdout": "", "stderr": "", "exit_code": 0}
15run_command
{
 "command": "cd /workspace && python3 - <<'PY'\nfrom pathlib import Path\np = Path(\"fastapi/sse.py\")\ns = p.read_text()\n\n# 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# I want to add newline check to it and create a similar one for 'event'.\n\n# Let's use a helper to avoid repetition.\n\n# I'll replace the _check_id_no_null function with a more comprehensive one and add another for event.\n\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\"\"\"\n\nnew = \"\"\"def _check_sse_field_no_newlines(v: str | None, field_name: str) -> str | None:\n    if v is not None and (\"\\\\n\" in v or \"\\\\r\" in v):\n        raise ValueError(f\"SSE '{field_name}' must not contain newline characters\")\n    return v\n\ndef _check_id_no_null(v: str | None) -> str | None:\n    if v is not None:\n        if \"\\\\0\" in v:\n            raise ValueError(\"SSE 'id' must not contain null characters\")\n        return _check_sse_field_no_newlines(v, \"id\")\n    return v\n\"\"\"\n# Wait, _check_sse_field_no_newlines is not used by AfterValidator because it takes 2 args.\n# AfterValidator takes a callable that takes 1 arg.\n\n# Let's just do it simply.\n\n# Actually, I can use a lambda or a partial, but it's cleaner to just have two functions or one that uses a closure.\n\n# Let's just use two functions.\n\n# Let's redefine _check_id_no_null and add _check_event_no_newlines.\nPY\n"
}
{"status": "ok", "stdout": "", "stderr": "", "exit_code": 0}
full
{"status": "ok", "stdout": "", "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\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\"\"\"\n\nnew = \"\"\"def _check_id_no_null(v: str | None) -> str | None:\n    if v is not None:\n        if \"\\\\0\" in v:\n            raise ValueError(\"SSE 'id' must not contain null characters\")\n        if \"\\\\n\" in v or \"\\\\r\" in v:\n            raise ValueError(\"SSE 'id' must not contain newline characters\")\n    return v\n\ndef _check_event_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 'event' must not contain newline characters\")\n    return v\n\"\"\"\np.write_text(s.replace(old, new))\nPY\n"
}
{"status": "ok", "stdout": "", "stderr": "", "exit_code": 0}
full
{"status": "ok", "stdout": "", "stderr": "", "exit_code": 0}
17run_command
{
 "command": "sed -n '36,50p' /workspace/fastapi/sse.py"
}
{"status": "ok", "stdout": "def _check_id_no_null(v: str | None) -> str | None:\n if v is not None:\n if \"\\0\" in v:\n raise ValueError(\"SSE 'id' must not co
full
{"status": "ok", "stdout": "def _check_id_no_null(v: str | None) -> str | None:\n    if v is not None:\n        if \"\\0\" in v:\n            raise ValueError(\"SSE 'id' must not contain null characters\")\n        if \"\\n\" in v or \"\\r\" in v:\n            raise ValueError(\"SSE 'id' must not contain newline characters\")\n    return v\n\ndef _check_event_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 'event' must not contain newline characters\")\n    return v\n\n\nclass ServerSentEvent(BaseModel):\n", "stderr": "", "exit_code": 0}
18run_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(\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\nnew = \"\"\"    event: Annotated[\n        str | None,\n        AfterValidator(_check_event_no_newlines),\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\"\"\"\np.write_text(s.replace(old, new))\nPY\n"
}
{"status": "ok", "stdout": "", "stderr": "", "exit_code": 0}
full
{"status": "ok", "stdout": "", "stderr": "", "exit_code": 0}
19run_command
{
 "command": "sed -n '88,102p' /workspace/fastapi/sse.py"
}
{"status": "ok", "stdout": " 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
full
{"status": "ok", "stdout": "            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        AfterValidator(_check_event_no_newlines),\n        Doc(\n            \"\"\"\n            Optional event type name.\n\n            Maps to `addEventListener(event, ...)` on the browser. When omitted,\n", "stderr": "", "exit_code": 0}
20run_command
{
 "command": "cd /workspace && python3 - <<'PY'\nfrom fastapi.sse import ServerSentEvent\nfrom pydantic import ValidationError\n\ntry:\n    ServerSentEvent(id=\"line1\\nline2\")\n    print(\"Failed: id with newline should raise ValidationError\")\nexcept ValidationError:\n    print(\"Passed: id with newline raised ValidationError\")\n\ntry:\n    ServerSentEvent(id=\"line1\\rline2\")\n    print(\"Failed: id with CRLF should raise ValidationError\")\nexcept ValidationError:\n    print(\"Passed: id with CRLF raised ValidationError\")\n\ntry:\n    ServerSentEvent(event=\"line1\\nline2\")\n    print(\"Failed: event with newline should raise ValidationError\")\nexcept ValidationError:\n    print(\"Passed: event with newline raised ValidationError\")\n\ntry:\n    ServerSentEvent(event=\"line1\\rline2\")\n    print(\"Failed: event with CRLF should raise ValidationError\")\nexcept ValidationError:\n    print(\"Passed: event with CRLF raised ValidationError\")\n\ntry:\n    ServerSentEvent(data=\"line1\\nline2\")\n    print(\"Passed: data with newline is allowed\")\nexcept ValidationError:\n    print(\"Failed: data with newline should NOT raise ValidationError\")\nPY\n"
}
{"status": "ok", "stdout": "Passed: id with newline raised ValidationError\nPassed: id with CRLF raised ValidationError\nPassed: event with newline raised Valid
full
{"status": "ok", "stdout": "Passed: id with newline raised ValidationError\nPassed: id with CRLF raised ValidationError\nPassed: event with newline raised ValidationError\nPassed: event with CRLF raised ValidationError\nPassed: data with newline is allowed\n", "stderr": "", "exit_code": 0}
21run_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.15s\n", "stderr": "", "exit_code": 0}
22run_command
{
 "command": "git diff"
}
{"status": "ok", "stdout": "diff --git a/fastapi/sse.py b/fastapi/sse.py\nindex 901d82496..b2f95a400 100644\n--- a/fastapi/sse.py\n+++ b/fastapi/sse.py\n@@ -34,
full
{"status": "ok", "stdout": "diff --git a/fastapi/sse.py b/fastapi/sse.py\nindex 901d82496..b2f95a400 100644\n--- a/fastapi/sse.py\n+++ b/fastapi/sse.py\n@@ -34,8 +34,16 @@ class EventSourceResponse(StreamingResponse):\n \n \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+    if v is not None:\n+        if \"\\0\" in v:\n+            raise ValueError(\"SSE 'id' must not contain null characters\")\n+        if \"\\n\" in v or \"\\r\" in v:\n+            raise ValueError(\"SSE 'id' must not contain newline characters\")\n+    return v\n+\n+def _check_event_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 'event' must not contain newline characters\")\n     return v\n \n \n@@ -86,6 +94,7 @@ class ServerSentEvent(BaseModel):\n     ] = None\n     event: Annotated[\n         str | None,\n+        AfterValidator(_check_event_no_newlines),\n         Doc(\n             \"\"\"\n             Optional event type name.\n", "stderr": "", "exit_code": 0}
23submit_patch
{}
{"status": "ok", "patch_size": 1040, "files_changed": 1}
full
{"status": "ok", "patch_size": 1040, "files_changed": 1}

Patch

diff --git a/fastapi/sse.py b/fastapi/sse.py
index 901d82496..b2f95a400 100644
--- a/fastapi/sse.py
+++ b/fastapi/sse.py
@@ -34,8 +34,16 @@ class EventSourceResponse(StreamingResponse):
 
 
 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")
+    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_newlines(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
 
 
@@ -86,6 +94,7 @@ class ServerSentEvent(BaseModel):
     ] = None
     event: Annotated[
         str | None,
+        AfterValidator(_check_event_no_newlines),
         Doc(
             """
             Optional event type name.

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.49s