Sanitize server-supplied strings at decoder boundary

FailureResponse.message, LeaderResponse.address, and
ServersResponse.nodes[*].address are decoded from server-supplied
bytes and flow directly into exception messages and log records. A
malicious or compromised peer could embed ANSI escape sequences to
corrupt operator terminals, CR/LF to forge log entries, or NUL bytes
that upset some log backends.

Replace C0 control characters (except tab 0x09 and LF 0x0A) and DEL
(0x7F) with the literal '?' in these three fields. CR is dropped
because it is the log-injection vector alongside LF, and LF alone is
enough to render legitimate multi-line server diagnostics. The
sanitization lives in a small _sanitize_server_text helper and is
called only from the three decode_body sites — SQL text, column
names, and other application strings are intentionally untouched.

Tests cover ANSI clear-screen, CR forging, raw NUL (hand-built body
to bypass encode_text's strict encoding), tab/LF preservation, and
all three message types.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: antoineleclair

src/dqlitewire/messages/responses.py

@@ -1,5 +1,6 @@
 """Server to client response messages."""
 
+import re
 from dataclasses import dataclass, field
 from typing import Any, ClassVar
 
@@ -39,6 +40,29 @@
 _MAX_FILE_COUNT = 100
 _MAX_NODE_COUNT = 10_000
 
+# Sanitize server-supplied text destined for exception messages and
+# logs. The C server promises UTF-8 but makes no promise about terminal
+# escapes or log-injection characters: a malicious or compromised peer
+# can embed ANSI colour/clear sequences, CR/LF to forge log lines, or
+# NUL bytes that upset some log backends. Replace C0 controls (except
+# tab 0x09 and LF 0x0A) and DEL (0x7F) with a literal "?". CR (0x0D) is
+# dropped — it is the log-injection vector alongside LF, and LF alone
+# is enough to represent legitimate multi-line server messages in
+# journald / file handlers.
+_CONTROL_CHARS_RE = re.compile(r"[\x00-\x08\x0b-\x1f\x7f]")
+
+
+def _sanitize_server_text(s: str) -> str:
+    """Replace C0 control characters and DEL with '?' in server strings.
+
+    Applied at the decoder boundary for text fields that flow directly
+    into exception messages and logs (FailureResponse.message,
+    LeaderResponse.address, ServersResponse.nodes[*].address). Leaves
+    tab and LF untouched so multi-line server diagnostics render
+    correctly.
+    """
+    return _CONTROL_CHARS_RE.sub("?", s)
+
 
 @dataclass
 class FailureResponse(Message):
@@ -66,7 +90,7 @@ def encode_body(self) -> bytes:
     def decode_body(cls, data: bytes, schema: int = 0) -> "FailureResponse":
         code = decode_uint64(data)
         message, _ = decode_text(data[8:])
-        return cls(code, message)
+        return cls(code, _sanitize_server_text(message))
 
 
 @dataclass
@@ -94,7 +118,7 @@ def decode_body(cls, data: bytes, schema: int = 0) -> "LeaderResponse":
         """
         node_id = decode_uint64(data)
         address, _ = decode_text(data[8:])
-        return cls(node_id, address)
+        return cls(node_id, _sanitize_server_text(address))
 
     @classmethod
     def decode_body_legacy(cls, data: bytes) -> "LeaderResponse":
@@ -104,7 +128,7 @@ def decode_body_legacy(cls, data: bytes) -> "LeaderResponse":
         Go reference: DecodeNodeLegacy in internal/protocol/message.go.
         """
         address, _ = decode_text(data)
-        return cls(node_id=0, address=address)
+        return cls(node_id=0, address=_sanitize_server_text(address))
 
 
 @dataclass
@@ -613,6 +637,7 @@ def decode_body(cls, data: bytes, schema: int = 0) -> "ServersResponse":
             node_id = decode_uint64(view[offset:])
             offset += 8
             address, consumed = decode_text(view[offset:])
+            address = _sanitize_server_text(address)
             offset += consumed
             raw_role = decode_uint64(view[offset:])
             offset += 8

tests/test_messages_responses.py

@@ -1311,6 +1311,81 @@ def test_leader_response_legacy_missing_null_terminator_raises(self) -> None:
             LeaderResponse.decode_body_legacy(b"abc")
 
 
+class TestServerTextSanitization:
+    """Server-supplied strings that flow into exception messages and
+    logs must have C0 control characters and DEL replaced with '?' so a
+    malicious server cannot inject ANSI escape sequences, CR/LF
+    log-forgery, or NUL bytes into operator-facing output.
+
+    Tab (0x09) and LF (0x0A) are preserved so multi-line diagnostics
+    render as real newlines. CR (0x0D) is dropped because it is the
+    log-injection vector.
+    """
+
+    def test_failure_response_sanitizes_ansi(self) -> None:
+        # Encode bytes directly because encode_text(str) stringifies.
+        from dqlitewire.types import encode_text, encode_uint64
+
+        payload = encode_uint64(42) + encode_text("\x1b[2J\x1b[Hwiped!")
+        decoded = FailureResponse.decode_body(payload)
+        assert decoded.message == "?[2J?[Hwiped!"
+
+    def test_failure_response_sanitizes_cr(self) -> None:
+        from dqlitewire.types import encode_text, encode_uint64
+
+        payload = encode_uint64(1) + encode_text("ok\r\nFAKE log line")
+        decoded = FailureResponse.decode_body(payload)
+        # LF preserved; CR replaced with '?' so it cannot forge a log line.
+        assert decoded.message == "ok?\nFAKE log line"
+
+    def test_failure_response_sanitizes_raw_nul(self) -> None:
+        # Construct a body whose text field contains a raw NUL byte
+        # *before* the real NUL terminator. The legitimate encoder
+        # rejects this, so hand-build the body to model a malicious
+        # peer that bypasses encode_text.
+        from dqlitewire.types import encode_uint64
+
+        # uint64 code (8 bytes) + raw "ab\x01cd" + NUL terminator + pad.
+        body = encode_uint64(1) + b"ab\x01cd\x00\x00\x00"
+        decoded = FailureResponse.decode_body(body)
+        assert decoded.message == "ab?cd"
+
+    def test_failure_response_preserves_tab_and_lf(self) -> None:
+        from dqlitewire.types import encode_text, encode_uint64
+
+        payload = encode_uint64(1) + encode_text("line1\nline2\tkey=val")
+        decoded = FailureResponse.decode_body(payload)
+        assert decoded.message == "line1\nline2\tkey=val"
+
+    def test_leader_response_sanitizes_address(self) -> None:
+        from dqlitewire.types import encode_text, encode_uint64
+
+        payload = encode_uint64(5) + encode_text("evil.com:9001\r\nHost: x")
+        decoded = LeaderResponse.decode_body(payload)
+        assert decoded.address == "evil.com:9001?\nHost: x"
+
+    def test_leader_response_legacy_sanitizes(self) -> None:
+        from dqlitewire.types import encode_text
+
+        payload = encode_text("\x1b[31mred\x1b[0m")
+        decoded = LeaderResponse.decode_body_legacy(payload)
+        assert decoded.address == "?[31mred?[0m"
+
+    def test_servers_response_sanitizes_node_addresses(self) -> None:
+        # Build a SERVERS body with a single node whose address contains
+        # an ANSI clear-screen sequence.
+        from dqlitewire.constants import NodeRole
+        from dqlitewire.types import encode_text, encode_uint64
+
+        body = encode_uint64(1)
+        body += encode_uint64(1)
+        body += encode_text("node1\x1b[2J:9001")
+        body += encode_uint64(NodeRole.VOTER)
+        decoded = ServersResponse.decode_body(body)
+        assert len(decoded.nodes) == 1
+        assert decoded.nodes[0].address == "node1?[2J:9001"
+
+
 class TestReservedFieldValidation:
     """Non-zero reserved bytes must fail to decode across all message
     types that declare a reserved field, matching LeaderRequest's