mirror of
https://github.com/kennethreitz/responder.git
synced 2026-07-21 18:39:29 +00:00
kennethreitz
4abd8bb8e9
## Summary
A runtime-hardening pass over the Server-Sent Events path (which the
earlier audit only skimmed) turned up a real injection gap in the SSE
frame encoder (`responder/models.py`).
`_sse_frame` interpolated the **single-line** fields — `event`, `id`,
`retry`, and the `comment` line — verbatim:
```python
lines.append(f"event: {event}") # before
```
SSE fields are terminated by a line break, so a `\r` or `\n` in any of
those values ends the field early and injects arbitrary SSE lines or
whole frames. That matters whenever those fields carry
caller/user-derived data — e.g. an event name, or an `id` echoed from a
client's `Last-Event-ID`. Separately, `data` only split on `\n`, so a
lone `\r` (a valid SSE line terminator per the spec) passed through and
could inject a new field.
## Fix
- Strip `\r`, `\n` (and `\x00`, which the spec forbids in `id`) from
`event`, `id`, `retry`, and comment values via a small
`_sse_single_line` helper.
- Normalize `data` on all line terminators (`\r\n`, `\r`, `\n`) before
splitting into `data:` lines, so multi-line data is encoded correctly
and no lone `\r` survives.
Legitimate multi-line `data` still works exactly as before.
## Tests
New `tests/test_sse_injection.py` (6 tests): newline stripped from
`event`/`id`/`retry`, NUL stripped from `id`, `data` split on all
terminators, comment scrubbed, and an end-to-end route asserting an
injected `data:` line can't appear. Existing `tests/test_sse.py` still
passes.
Full suite: **755 passed**, 1 skipped (php not installed); ruff clean;
mypy clean.
## Scope note
I checked the other runtime paths in the same neighborhood and found
them already solid, so this PR is deliberately narrow rather than a
grab-bag: request-body size limits are enforced while streaming (both
declared `Content-Length` and accumulated bytes, 413 on exceed);
streaming/response disconnect is delegated to Starlette's
`StreamingResponse`; SSE already sets `Cache-Control: no-cache` /
`X-Accel-Buffering: no` / keep-alive and supports a heartbeat;
open-redirects are guarded by `allow_external`.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
67 lines
2.2 KiB
Python
67 lines
2.2 KiB
Python
"""SSE frames must not be injectable via newlines in caller-supplied fields.
|
|
|
|
``event``/``id``/``retry``/comment are single-line fields; a CR/LF there would
|
|
terminate the field and let the value inject extra SSE lines or frames. ``data``
|
|
must split on any line terminator (\\r\\n, \\r, \\n) into multiple ``data:``
|
|
lines rather than pass a lone \\r through.
|
|
"""
|
|
|
|
from starlette.testclient import TestClient
|
|
|
|
import responder
|
|
from responder.models import _format_sse_event, _sse_frame
|
|
|
|
|
|
def _client(api):
|
|
return TestClient(api, base_url="http://;")
|
|
|
|
|
|
def test_newline_in_event_field_is_stripped():
|
|
frame = _sse_frame(data="ok", event="tick\ndata: injected\n\nevent: evil").decode()
|
|
# The event value's newline must not create a new line/frame.
|
|
assert "event: tickdata: injected" in frame
|
|
assert "\ndata: injected" not in frame
|
|
assert "\n\nevent: evil" not in frame
|
|
|
|
|
|
def test_newline_in_id_and_retry_is_stripped():
|
|
frame = _sse_frame(data="x", id="1\nevent: nope", retry="5\ndata: nope").decode()
|
|
assert "id: 1event: nope" in frame
|
|
assert "retry: 5data: nope" in frame
|
|
# Exactly one blank-line frame terminator, so no extra frame was injected.
|
|
assert frame.count("\n\n") == 1
|
|
|
|
|
|
def test_nul_in_id_is_stripped():
|
|
# The SSE spec forbids NUL in the id field.
|
|
frame = _sse_frame(data="x", id="a\x00b").decode()
|
|
assert "id: ab" in frame
|
|
assert "\x00" not in frame
|
|
|
|
|
|
def test_data_splits_on_all_line_terminators():
|
|
frame = _sse_frame(data="a\r\nb\rc\nd").decode()
|
|
assert "data: a\ndata: b\ndata: c\ndata: d" in frame
|
|
# No stray carriage return survives inside a data line.
|
|
assert "\r" not in frame
|
|
|
|
|
|
def test_comment_newline_is_stripped():
|
|
frame = _format_sse_event({"comment": "keep\ndata: injected"}).decode()
|
|
assert frame == ": keepdata: injected\n\n"
|
|
|
|
|
|
def test_end_to_end_event_injection_blocked():
|
|
api = responder.API(allowed_hosts=[";"], session_https_only=False)
|
|
|
|
@api.route("/s")
|
|
async def s(req, resp):
|
|
@resp.sse
|
|
async def stream():
|
|
yield {"data": "hello", "event": "tick\ndata: 9999"}
|
|
|
|
body = _client(api).get("/s").text
|
|
assert "event: tickdata: 9999" in body
|
|
# The injected "data: 9999" must not appear as its own line.
|
|
assert "\ndata: 9999" not in body
|