diff --git a/docs/project/changelog.rst b/docs/project/changelog.rst index 3e147f850..c670792fe 100644 --- a/docs/project/changelog.rst +++ b/docs/project/changelog.rst @@ -50,6 +50,9 @@ Bug fixes * Fixed a regression from 16.1 where the legacy implementation rejected non-ASCII headers. +* Escaped non-ASCII text in PING and PONG frame log messages to avoid errors + with non-UTF-8 log handlers. + .. _17.1: 17.1 diff --git a/src/websockets/frames.py b/src/websockets/frames.py index 318bb8228..fe3dabb45 100644 --- a/src/websockets/frames.py +++ b/src/websockets/frames.py @@ -193,6 +193,12 @@ def _data_repr(self) -> tuple[str, bool | None]: # display UTF-8 text in binary frames nicely and generally to be helpful # and robust. Also support frames fragmented within UTF-8 sequences. + def repr_text(data: bytes) -> str: + decoded = data.decode(errors="replace") + # Ping and pong payloads may be generated internally. Escape + # non-ASCII characters so logging them is safe for any encoding. + return ascii(decoded) if self.opcode in (PING, PONG) else repr(decoded) + if len(self.data) > 4 * self.MAX_LOG_SIZE: # Process only the start and the end, as the middle will be elided. # Cast to bytes because self.data could be a memoryview. @@ -206,7 +212,7 @@ def _data_repr(self) -> tuple[str, bool | None]: must_end_clean=self.fin, ) if is_text: - data_repr = repr((data_start + data_end).decode(errors="replace")) + data_repr = repr_text(data_start + data_end) else: # Cast to bytes because self.data could be a memoryview. @@ -217,7 +223,7 @@ def _data_repr(self) -> tuple[str, bool | None]: must_end_clean=self.fin, ) if is_text: - data_repr = repr(data.decode(errors="replace")) + data_repr = repr_text(data) # When the payload is text (except perhaps for boundaries), we decoded # enough in ``data_repr``. Now, do the same when the payload is binary. diff --git a/tests/test_frames.py b/tests/test_frames.py index 815e6bd14..5bc7212d6 100644 --- a/tests/test_frames.py +++ b/tests/test_frames.py @@ -364,6 +364,14 @@ def test_ping_text(self): "PING 'ping' [text, 4 bytes]", ) + def test_ping_pong_text_is_ascii(self): + for opcode in (PING, PONG): + with self.subTest(opcode=opcode): + self.assertEqual( + str(Frame(opcode, b"F\xd6\x8a}")), + f"{opcode.name} " + "'F\\u058a}' [text, 4 bytes]", + ) + def test_ping_text_with_newline(self): self.assertEqual( str(Frame(PING, b"ping\n")),