fix: poison buffer after capped oversized-message skip

When skip_message() caps the skip to max_message_size for an oversized
message, the peer actually sent more bytes than were discarded. The
stream is permanently desynchronized — the remaining undiscarded bytes
would be interpreted as a new message header, producing silent garbage.

Now the buffer is poisoned (via deferred _poison_after_skip flag) once
the capped skip completes, forcing callers to reset() and reconnect.
Previously skip_message() returned True as if recovery succeeded.

Closes #121

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

Author: antoineleclair

src/dqlitewire/buffer.py

@@ -94,6 +94,7 @@ def __init__(self, max_message_size: int = DEFAULT_MAX_MESSAGE_SIZE) -> None:
         self._max_message_size = max_message_size
         self._skip_remaining = 0
         self._poisoned: Exception | None = None
+        self._poison_after_skip: Exception | None = None
 
     @property
     def is_poisoned(self) -> bool:
@@ -117,6 +118,7 @@ def reset(self) -> None:
         self._pos = 0
         self._skip_remaining = 0
         self._poisoned = None
+        self._poison_after_skip = None
 
     def _check_poisoned(self) -> None:
         if self._poisoned is not None:
@@ -182,6 +184,12 @@ def feed(self, data: bytes) -> None:
                 discard = min(len(data), self._skip_remaining)
                 data = data[discard:]
                 self._skip_remaining -= discard
+                # Trigger deferred poison once the capped skip completes.
+                # The stream is desynchronized — remaining peer bytes were
+                # not discarded — so all further reads would be garbage.
+                if self._skip_remaining == 0 and self._poison_after_skip is not None:
+                    self._poisoned = self._poison_after_skip
+                    self._poison_after_skip = None
                 if not data:
                     return
             self._maybe_compact()
@@ -358,6 +366,27 @@ def skip_message(self) -> bool:
                 skip_now = min(effective_total, available)
                 self._pos += skip_now
                 self._skip_remaining = effective_total - skip_now
+                # If the cap is smaller than the actual message, mark for
+                # deferred poisoning. We cannot poison immediately because
+                # feed() needs to keep discarding bytes during the skip.
+                # Once _skip_remaining reaches 0, feed() will poison.
+                if effective_total < total_size:
+                    if self._skip_remaining == 0:
+                        # Capped skip completed immediately — poison now.
+                        self.poison(
+                            DecodeError(
+                                f"Oversized message skip capped to "
+                                f"{effective_total} of {total_size} bytes; "
+                                f"stream is desynchronized. Call reset()."
+                            )
+                        )
+                    else:
+                        # Deferred: feed() will poison once skip completes.
+                        self._poison_after_skip = DecodeError(
+                            f"Oversized message skip capped to "
+                            f"{effective_total} of {total_size} bytes; "
+                            f"stream is desynchronized. Call reset()."
+                        )
             self._maybe_compact()
         except BaseException as e:
             if self._poisoned is None:

tests/test_buffer.py

@@ -413,24 +413,24 @@ def test_rejects_oversized_message(self) -> None:
         with pytest.raises(DecodeError, match="exceeds maximum"):
             buf.read_message()
 
-    def test_skip_message_recovers_from_oversized(self) -> None:
-        """After skipping an oversized message, stream should not be corrupted.
+    def test_skip_message_poisons_after_capped_oversized(self) -> None:
+        """After a capped oversized skip, stream is desynchronized and
+        the buffer must be poisoned (issue 121).
 
         skip_message caps _skip_remaining to max_message_size to prevent
-        amplification attacks. The test feeds at most max_message_size bytes
-        of oversized body, then verifies the next message decodes correctly.
+        amplification attacks. Since the peer sent more bytes than the cap,
+        the stream is permanently desynchronized. The buffer is poisoned
+        once the capped skip completes.
         """
         import struct
 
         import pytest
 
-        from dqlitewire.exceptions import DecodeError
+        from dqlitewire.exceptions import DecodeError, ProtocolError
 
         buf = ReadBuffer(max_message_size=1024)
         # Header claiming a huge body: size_words=1000 (8000 bytes > 1024 limit)
         oversized_header = struct.pack("<IBBH", 1000, 0, 0, 0)
-        valid_msg = LeaderResponse(node_id=1, address="node1:9001")
-        valid_encoded = valid_msg.encode()
 
         # Feed header + first chunk
         buf.feed(oversized_header + b"\xab" * 500)
@@ -443,25 +443,60 @@ def test_skip_message_recovers_from_oversized(self) -> None:
         # skip_message should return False — partial skip
         assert buf.skip_message() is False
 
-        # Feed enough bytes to complete the capped skip + valid message.
-        # _skip_remaining is capped to max_message_size (1024), and skip_message
-        # already consumed the 508 bytes in the buffer (8 header + 500 body),
-        # so _skip_remaining = 1024 - 508 = 516 bytes.
-        buf.feed(b"\xab" * 516 + valid_encoded)
+        # Feed enough bytes to complete the capped skip.
+        remaining = buf._skip_remaining
+        buf.feed(b"\xab" * remaining)
 
-        # The capped oversized bytes should be discarded; valid message readable
-        assert buf.has_message()
-        data = buf.read_message()
-        assert data is not None
-        assert data == valid_encoded
+        # Skip is complete but buffer is poisoned — stream is desynchronized
+        assert not buf.is_skipping
+        assert buf.is_poisoned
+
+        with pytest.raises(ProtocolError, match="poisoned"):
+            buf.read_message()
 
-    def test_single_feed_completes_skip_and_appends_message(self) -> None:
-        """118: single feed() that completes skip AND contains next valid message."""
+    def test_capped_oversized_skip_poisons_buffer(self) -> None:
+        """121: capped skip of oversized message must poison the buffer.
+
+        When the header claims a body larger than max_message_size, the skip
+        is capped to max_message_size bytes. Since the peer sent more bytes
+        than were discarded, the stream is permanently desynchronized. The
+        buffer must be poisoned to prevent silent garbage reads.
+        """
         import struct
 
         import pytest
 
-        from dqlitewire.exceptions import DecodeError
+        from dqlitewire.exceptions import DecodeError, ProtocolError
+
+        buf = ReadBuffer(max_message_size=64)
+        # Header claiming 200 words = 1600 bytes body (>> 64 byte limit)
+        oversized_header = struct.pack("<IBBH", 200, 0, 0, 0)
+        buf.feed(oversized_header)
+
+        assert buf.has_message() is True
+        with pytest.raises(DecodeError, match="exceeds maximum"):
+            buf.read_message()
+
+        # skip_message triggers capped skip
+        buf.skip_message()
+
+        # After the capped skip completes, buffer must be poisoned
+        remaining = buf._skip_remaining
+        buf.feed(b"\x00" * remaining)
+        assert not buf.is_skipping
+        assert buf.is_poisoned
+
+        # Further operations should raise ProtocolError
+        with pytest.raises(ProtocolError, match="poisoned"):
+            buf.read_message()
+
+    def test_single_feed_completes_capped_skip_and_poisons(self) -> None:
+        """118/121: single feed() that completes capped skip poisons buffer."""
+        import struct
+
+        import pytest
+
+        from dqlitewire.exceptions import DecodeError, ProtocolError
 
         buf = ReadBuffer(max_message_size=64)
         # 200 words = 1600 bytes body, well over 64-byte limit
@@ -474,37 +509,34 @@ def test_single_feed_completes_skip_and_appends_message(self) -> None:
         assert buf.skip_message() is False
         assert buf.is_skipping
 
-        # Build a single payload: remaining skip bytes + valid message
+        # Build a single payload: remaining skip bytes + extra data
         remaining = buf._skip_remaining
-        valid_msg = LeaderRequest()
-        valid_encoded = valid_msg.encode()
-        combined = b"\xcc" * remaining + valid_encoded
+        combined = b"\xcc" * remaining + b"\x00" * 16
 
         buf.feed(combined)
 
         assert not buf.is_skipping
-        assert buf.has_message()
-        data = buf.read_message()
-        assert data is not None
-        assert data == valid_encoded
+        assert buf.is_poisoned
+
+        with pytest.raises(ProtocolError, match="poisoned"):
+            buf.read_message()
 
-    def test_skip_oversized_across_multiple_feeds(self) -> None:
-        """Oversized message bytes should be discarded across multiple feed() calls.
+    def test_skip_oversized_across_multiple_feeds_poisons(self) -> None:
+        """Oversized skip across multiple feeds poisons when complete (issue 121).
 
         skip_message caps _skip_remaining to max_message_size, so only that
-        many bytes are discarded (not the full claimed body size).
+        many bytes are discarded. Once the capped skip completes, the buffer
+        is poisoned because the stream is desynchronized.
         """
         import struct
 
         import pytest
 
-        from dqlitewire.exceptions import DecodeError
+        from dqlitewire.exceptions import DecodeError, ProtocolError
 
         buf = ReadBuffer(max_message_size=64)
         # 200 words = 1600 bytes body, well over 64-byte limit
         oversized_header = struct.pack("<IBBH", 200, 0, 0, 0)
-        valid_msg = LeaderRequest()
-        valid_encoded = valid_msg.encode()
 
         # Feed just the header
         buf.feed(oversized_header)
@@ -513,20 +545,20 @@ def test_skip_oversized_across_multiple_feeds(self) -> None:
             buf.read_message()
         assert buf.skip_message() is False
 
-        # Feed capped body in chunks (max_message_size=64, header was 8 bytes,
-        # so _skip_remaining = 64 - 8 = 56 bytes)
+        # Feed capped body in chunks
         remaining = buf._skip_remaining
         body = b"\xcc" * remaining
         while body:
             chunk = body[:20]
             body = body[20:]
             buf.feed(chunk)
 
-        # Now feed the valid message
-        buf.feed(valid_encoded)
-        assert buf.has_message()
-        data = buf.read_message()
-        assert data == valid_encoded
+        # Capped skip is complete — buffer should be poisoned
+        assert not buf.is_skipping
+        assert buf.is_poisoned
+
+        with pytest.raises(ProtocolError, match="poisoned"):
+            buf.feed(b"\x00" * 16)
 
     def test_is_skipping_property(self) -> None:
         """is_skipping reflects whether an oversized skip is in progress."""

tests/test_codec.py

@@ -1243,44 +1243,40 @@ def test_skip_message_exists_on_decoder(self) -> None:
         assert hasattr(decoder, "skip_message")
         assert hasattr(decoder, "is_skipping")
 
-    def test_skip_oversized_and_recover(self) -> None:
-        """After skipping an oversized message, normal messages should decode."""
+    def test_skip_oversized_poisons_after_capped_skip(self) -> None:
+        """After a capped oversized skip, buffer is poisoned (issue 121)."""
         import struct
 
-        from dqlitewire.exceptions import DecodeError
+        from dqlitewire.exceptions import DecodeError, ProtocolError
 
         # Create a decoder with a small max_message_size
         decoder = MessageDecoder(is_request=False)
         decoder._buffer._max_message_size = 128
 
         # Build an oversized message header (size_words=100 → 808 bytes total)
         oversized_header = struct.pack("<IBBH", 100, 0, 0, 0)
-        # Build a normal message after it
-        normal_msg = FailureResponse(code=42, message="test").encode()
 
         # Feed just the oversized header
         decoder.feed(oversized_header)
 
-        # has_message() is total; the raise surfaces from decode() / read_message().
         assert decoder.has_message() is True
         with pytest.raises(DecodeError, match="exceeds maximum"):
             decoder.decode()
 
         # skip_message() should handle the oversized message
         result = decoder.skip_message()
-        assert result is False  # haven't received full oversized body yet
+        assert result is False
         assert decoder.is_skipping is True
 
-        # Feed enough bytes to complete the capped skip + normal message.
-        # _skip_remaining is capped to max_message_size (128), header was 8,
-        # so _skip_remaining = 128 - 8 = 120 bytes.
-        decoder.feed(b"\x00" * decoder._buffer._skip_remaining + normal_msg)
+        # Feed enough bytes to complete the capped skip
+        decoder.feed(b"\x00" * decoder._buffer._skip_remaining)
 
-        # Now should be able to decode the normal message
+        # Buffer is now poisoned — stream is desynchronized
         assert decoder.is_skipping is False
-        decoded = decoder.decode()
-        assert isinstance(decoded, FailureResponse)
-        assert decoded.code == 42
+        assert decoder.is_poisoned is True
+
+        with pytest.raises(ProtocolError, match="poisoned"):
+            decoder.decode()
 
 
 class TestDecoderPoisonedState:
@@ -1348,10 +1344,13 @@ def test_reset_unpoisons_decoder(self) -> None:
         with pytest.raises(ProtocolError):
             decoder.decode()
 
-    def test_oversized_header_does_not_poison(self) -> None:
+    def test_oversized_header_does_not_poison_before_skip(self) -> None:
         """An oversized-header error from the buffer framing layer is
         recoverable via skip_message(); it must NOT poison because the
         bytes have not yet been consumed from the buffer.
+
+        After a capped skip completes, the buffer IS poisoned because the
+        stream is desynchronized (issue 121).
         """
         import struct
 
@@ -1362,18 +1361,13 @@ def test_oversized_header_does_not_poison(self) -> None:
 
         with pytest.raises(DecodeError, match="exceeds maximum"):
             decoder.decode()
-        # NOT poisoned — skip_message() is the documented recovery path.
+        # NOT poisoned before skip — skip_message() is the recovery path.
         assert decoder.is_poisoned is False
 
-        # Recovery via skip_message should still work.
+        # After capped skip completes, buffer IS poisoned (desync).
         decoder.skip_message()
         decoder.feed(b"\x00" * decoder._buffer._skip_remaining)
-        assert decoder.is_poisoned is False
-
-        # And a fresh message decodes afterwards.
-        normal = FailureResponse(code=1, message="ok").encode()
-        decoder.feed(normal)
-        assert isinstance(decoder.decode(), FailureResponse)
+        assert decoder.is_poisoned is True
 
     def test_poison_catches_non_protocol_exceptions(self) -> None:
         """Any exception from decode_bytes — not just DecodeError/ProtocolError —