From a9e7deccdc744da1ac2427b85856c1277f090ea1 Mon Sep 17 00:00:00 2001
From: Leul Tewodros Agonafer <107800362+LeulTew@users.noreply.github.com>
Date: Wed, 16 Sep 2026 22:54:10 +0300
Subject: [PATCH] test: guard incremental receive-buffer scanning

Bound ranges passed to real searches while receiving fragmented lines and header blocks, including repeated polls and prefix consumption.

Refs #16.
---
 h11/tests/test_receivebuffer.py | 111 +++++++++++++++++++++++++++++++-
 1 file changed, 110 insertions(+), 1 deletion(-)

diff --git a/h11/tests/test_receivebuffer.py b/h11/tests/test_receivebuffer.py
index 21a3870..47d4df1 100644
--- a/h11/tests/test_receivebuffer.py
+++ b/h11/tests/test_receivebuffer.py
@@ -1,8 +1,10 @@
 import re
-from typing import Tuple
+from types import SimpleNamespace
+from typing import List, Match, Optional, Tuple
 
 import pytest
 
+from .. import _receivebuffer
 from .._receivebuffer import ReceiveBuffer
 
 
@@ -133,3 +135,110 @@ def test_receivebuffer_for_invalid_delimiter(data: Tuple[bytes]) -> None:
         b"Connection: close",
     ]
     assert bytes(b) == b"Some body"
+
+
+@pytest.fixture
+def receivebuffer_with_scan_sizes(
+    monkeypatch: pytest.MonkeyPatch,
+) -> Tuple[ReceiveBuffer, List[int]]:
+    """Count search-window sizes, not timings or native bytearray copying."""
+
+    # A subclass lets us wrap find on this instance without replacing its storage.
+    class TrackedBytearray(bytearray):
+        pass
+
+    data = TrackedBytearray()
+    scan_sizes: List[int] = []
+    real_find = data.find
+    real_search = _receivebuffer.blank_line_regex.search
+
+    def find(sub: bytes, start: int = 0) -> int:
+        scan_sizes.append(max(0, len(data) - start))
+        return real_find(sub, start)
+
+    def search(buffer: bytearray, start: int = 0) -> Optional[Match[bytes]]:
+        scan_sizes.append(max(0, len(buffer) - start))
+        return real_search(buffer, start)
+
+    monkeypatch.setattr(data, "find", find)
+    monkeypatch.setattr(
+        _receivebuffer, "blank_line_regex", SimpleNamespace(search=search)
+    )
+    b = ReceiveBuffer()
+    b._data = data
+    return b, scan_sizes
+
+
+@pytest.mark.parametrize("size", [256, 4096])
+@pytest.mark.parametrize("consume", [0, 32])
+def test_receivebuffer_next_line_scan_bounds(
+    receivebuffer_with_scan_sizes: Tuple[ReceiveBuffer, List[int]],
+    size: int,
+    consume: int,
+) -> None:
+    b, scan_sizes = receivebuffer_with_scan_sizes
+    line = b"x" * size + b"\r\n"
+    following = b"next\r\n"
+
+    for _ in range(3):
+        scan_sizes.clear()
+        for byte in line[:-1]:
+            b += bytes([byte])
+            assert b.maybe_extract_next_line() is None
+            assert b.maybe_extract_next_line() is None
+
+        if consume:
+            assert b.maybe_extract_at_most(consume) == line[:consume]
+
+        b += line[-1:] + following
+        assert b.maybe_extract_next_line() == line[consume:]
+        assert b.maybe_extract_next_line() == following
+        assert not b
+
+        # Allow delimiter overlap, unchanged-buffer polls, and a rescan after
+        # consuming a prefix, but not a full-buffer scan for every new byte.
+        assert scan_sizes
+        assert sum(scan_sizes) <= 8 * (len(line) + len(following))
+
+
+@pytest.mark.parametrize("size", [256, 4096])
+@pytest.mark.parametrize("consume", [0, 32])
+@pytest.mark.parametrize(
+    "line_ending,blank_line",
+    [
+        pytest.param(b"\r\n", b"\r\n", id="crlf"),
+        pytest.param(b"\n", b"\n", id="lf"),
+        pytest.param(b"\n", b"\r\n", id="lf-crlf"),
+        pytest.param(b"\r\n", b"\n", id="crlf-lf"),
+    ],
+)
+def test_receivebuffer_lines_scan_bounds(
+    receivebuffer_with_scan_sizes: Tuple[ReceiveBuffer, List[int]],
+    size: int,
+    consume: int,
+    line_ending: bytes,
+    blank_line: bytes,
+) -> None:
+    b, scan_sizes = receivebuffer_with_scan_sizes
+    lines = [b"first: " + b"x" * size, b"second: value"]
+    block = line_ending.join(lines) + line_ending + blank_line
+    following = b"next: value\r\n\r\n"
+
+    for _ in range(3):
+        scan_sizes.clear()
+        for byte in block[:-1]:
+            b += bytes([byte])
+            assert b.maybe_extract_lines() is None
+            assert b.maybe_extract_lines() is None
+
+        if consume:
+            assert b.maybe_extract_at_most(consume) == block[:consume]
+
+        b += block[-1:] + following + blank_line
+        assert b.maybe_extract_lines() == [lines[0][consume:], lines[1]]
+        assert b.maybe_extract_lines() == [b"next: value"]
+        assert b.maybe_extract_lines() == []
+        assert not b
+
+        assert scan_sizes
+        assert sum(scan_sizes) <= 8 * (len(block) + len(following) + len(blank_line))