686f6c61
0e5ff99aeb
Starlette encodes send_text as UTF-8. A lone UTF-16 surrogate in a status payload used to UnicodeEncodeError and latch the whole Desktop connection closed. Sanitize the frame and skip only that send.
64 lines
2.0 KiB
Python
64 lines
2.0 KiB
Python
"""Lone UTF-16 surrogates must not tear down the Desktop WebSocket (#97288)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
|
|
from tui_gateway.ws import WSTransport, _sanitize_ws_text
|
|
|
|
|
|
LONE_SURROGATE = "\ud83d"
|
|
|
|
|
|
def test_sanitize_ws_text_makes_utf8_encodable() -> None:
|
|
dirty = f"gateway.ready {LONE_SURROGATE} payload"
|
|
out = _sanitize_ws_text(dirty)
|
|
out.encode("utf-8")
|
|
assert LONE_SURROGATE not in out
|
|
|
|
|
|
def test_sanitize_ws_text_leaves_valid_text_unchanged() -> None:
|
|
clean = '{"type":"gateway.ready","ok":true}'
|
|
assert _sanitize_ws_text(clean) is clean or _sanitize_ws_text(clean) == clean
|
|
|
|
|
|
class _FakeWS:
|
|
def __init__(self) -> None:
|
|
self.sent: list[str] = []
|
|
self.raise_on: str | None = None
|
|
|
|
async def send_text(self, line: str) -> None:
|
|
line.encode("utf-8")
|
|
if self.raise_on is not None and self.raise_on in line:
|
|
raise UnicodeEncodeError("utf-8", line, 0, 1, "surrogates not allowed")
|
|
self.sent.append(line)
|
|
|
|
|
|
def test_safe_send_sanitizes_surrogate_and_keeps_connection() -> None:
|
|
async def _run() -> None:
|
|
loop = asyncio.get_running_loop()
|
|
ws = _FakeWS()
|
|
transport = WSTransport(ws, loop, peer="127.0.0.1:1")
|
|
dirty = f'{{"type":"gateway.ready","x":"{LONE_SURROGATE}"}}'
|
|
await transport._safe_send_many(["first", dirty, "third"])
|
|
assert transport.closed is False
|
|
assert ws.sent[0] == "first"
|
|
assert ws.sent[-1] == "third"
|
|
assert LONE_SURROGATE not in "".join(ws.sent)
|
|
assert len(ws.sent) == 3
|
|
|
|
asyncio.run(_run())
|
|
|
|
|
|
def test_unicode_encode_error_does_not_close_socket() -> None:
|
|
async def _run() -> None:
|
|
loop = asyncio.get_running_loop()
|
|
ws = _FakeWS()
|
|
ws.raise_on = "BOOM"
|
|
transport = WSTransport(ws, loop, peer="127.0.0.1:1")
|
|
await transport._safe_send_many(["ok-a", "BOOM-frame", "ok-b"])
|
|
assert transport.closed is False
|
|
assert ws.sent == ["ok-a", "ok-b"]
|
|
|
|
asyncio.run(_run())
|