fix: bound stdin reads to 10 MiB via sys.stdin.buffer for byte-accurate limiting

Address Copilot review feedback on PR #3903:

- Use sys.stdin.buffer.read() instead of sys.stdin.read() so the 10 MiB
  limit is enforced on raw bytes rather than Unicode code points (a 4-byte
  UTF-8 sequence now counts as 4 bytes, not 1 character).
- Add regression tests for both the events module and CLI event runner:
  below-limit, exact-limit, oversized, multibyte UTF-8, empty stdin,
  TTY, and invalid UTF-8 replacement.

Author: Quratulain-bilal

src/specify_cli/commands/event.py

@@ -6,6 +6,29 @@
 import sys
 import typer
 
+_MAX_STDIN_BYTES = 10 * 1024 * 1024  # 10 MiB
+
+
+def _read_stdin_bounded(max_bytes: int = _MAX_STDIN_BYTES) -> str:
+    """Read at most *max_bytes* from stdin to prevent unbounded memory use.
+
+    Uses ``sys.stdin.buffer`` so the limit is enforced on raw bytes rather
+    than Unicode code points — a 4-byte UTF-8 sequence counts as 4 bytes,
+    not 1 character.
+    """
+    if sys.stdin.isatty():
+        return "{}"
+    chunks: list[bytes] = []
+    total = 0
+    while total < max_bytes:
+        chunk = sys.stdin.buffer.read(min(max_bytes - total, 65536))
+        if not chunk:
+            break
+        chunks.append(chunk)
+        total += len(chunk)
+    return b"".join(chunks).decode("utf-8", errors="replace")
+
+
 event_app = typer.Typer(
     name="event",
     help="Manage and execute event-driven commands",
@@ -24,19 +47,7 @@ def event_run(
     """Resolve and run an event-driven command script with stdin payload."""
     from ..events import resolve_and_run_event_command
 
-    # Read payload from stdin if available (capped at 1 MiB to prevent DoS).
-    MAX_STDIN_BYTES = 1 * 1024 * 1024
-    if not sys.stdin.isatty():
-        raw = sys.stdin.read(MAX_STDIN_BYTES)
-        if not sys.stdin.eof:
-            raise typer.Exit(
-                code=1,
-                message="stdin payload exceeds 1 MiB limit; "
-                "truncate or pipe a smaller payload",
-            )
-        payload = raw
-    else:
-        payload = "{}"
+    payload = _read_stdin_bounded()
 
     # Run the event command
     project_root = Path.cwd()  # The agent runs events from project root

src/specify_cli/events.py

@@ -61,6 +61,30 @@
     "stop",
 })
 
+# -- Stdin bounded read ---------------------------------------------------
+
+_MAX_STDIN_BYTES = 10 * 1024 * 1024  # 10 MiB
+
+
+def _read_stdin_bounded(max_bytes: int = _MAX_STDIN_BYTES) -> str:
+    """Read at most *max_bytes* from stdin to prevent unbounded memory use.
+
+    Uses ``sys.stdin.buffer`` so the limit is enforced on raw bytes rather
+    than Unicode code points — a 4-byte UTF-8 sequence counts as 4 bytes,
+    not 1 character.
+    """
+    if sys.stdin.isatty():
+        return "{}"
+    chunks: list[bytes] = []
+    total = 0
+    while total < max_bytes:
+        chunk = sys.stdin.buffer.read(min(max_bytes - total, 65536))
+        if not chunk:
+            break
+        chunks.append(chunk)
+        total += len(chunk)
+    return b"".join(chunks).decode("utf-8", errors="replace")
+
 # -- Events Dispatcher template ---------------------------------------------
 
 _EVENTS_DISPATCHER_TEMPLATE = '''#!/usr/bin/env python3
@@ -347,7 +371,7 @@ def main():
     # hookEventName field (required by Qwen's hooks spec; included by
     # Gemini/Tabnine/Devin which derive from the same protocol).
     native_event = sys.argv[5] if len(sys.argv) >= 6 else ""
-    payload = sys.stdin.read() if not sys.stdin.isatty() else "{}"
+    payload = _read_stdin_bounded()
     project_root = Path(__file__).parent.parent.resolve()
 
     # Preferred path: specify_cli is importable (durable install) — delegate to

tests/integrations/test_events.py

@@ -2525,3 +2525,151 @@ def test_refresh_failure_preserves_existing_config(self, tmp_path):
         # The pre-existing config was NOT destroyed before the failure
         # (install handles cleanup atomically; refresh no longer pre-strips).
         assert config_path.read_text() == original
+
+
+# -- Bounded stdin reader ---------------------------------------------------
+
+class TestReadStdinBounded:
+    """Test _read_stdin_bounded byte-accurate limiting."""
+
+    def test_small_payload_passes_through(self):
+        from specify_cli.events import _read_stdin_bounded
+        from io import BytesIO
+
+        buf = BytesIO(b'{"key": "value"}')
+        stdin_mock = MagicMock()
+        stdin_mock.isatty.return_value = False
+        stdin_mock.buffer = buf
+        with patch("specify_cli.events.sys") as mock_sys:
+            mock_sys.stdin = stdin_mock
+            result = _read_stdin_bounded(max_bytes=1024)
+        assert result == '{"key": "value"}'
+
+    def test_multibyte_utf8_counted_as_bytes(self):
+        """A 4-byte UTF-8 character counts as 4 bytes, not 1 character."""
+        from specify_cli.events import _read_stdin_bounded
+        from io import BytesIO
+
+        # U+1F600 (😀) is 4 bytes in UTF-8
+        payload = "hello \U0001F600 world".encode("utf-8")
+        assert len(payload) == 16  # "hello " (6) + 😀 (4) + " world" (6)
+
+        buf = BytesIO(payload)
+        stdin_mock = MagicMock()
+        stdin_mock.isatty.return_value = False
+        stdin_mock.buffer = buf
+        with patch("specify_cli.events.sys") as mock_sys:
+            mock_sys.stdin = stdin_mock
+            result = _read_stdin_bounded(max_bytes=16)
+        assert len(result.encode("utf-8")) == 16
+
+    def test_oversized_payload_truncated(self):
+        from specify_cli.events import _read_stdin_bounded
+        from io import BytesIO
+
+        buf = BytesIO(b"x" * 200)
+        stdin_mock = MagicMock()
+        stdin_mock.isatty.return_value = False
+        stdin_mock.buffer = buf
+        with patch("specify_cli.events.sys") as mock_sys:
+            mock_sys.stdin = stdin_mock
+            result = _read_stdin_bounded(max_bytes=100)
+        assert len(result.encode("utf-8")) == 100
+        assert result == "x" * 100
+
+    def test_exact_limit_passes(self):
+        from specify_cli.events import _read_stdin_bounded
+        from io import BytesIO
+
+        buf = BytesIO(b"a" * 65536)
+        stdin_mock = MagicMock()
+        stdin_mock.isatty.return_value = False
+        stdin_mock.buffer = buf
+        with patch("specify_cli.events.sys") as mock_sys:
+            mock_sys.stdin = stdin_mock
+            result = _read_stdin_bounded(max_bytes=65536)
+        assert result == "a" * 65536
+
+    def test_tty_returns_empty_json(self):
+        from specify_cli.events import _read_stdin_bounded
+
+        stdin_mock = MagicMock()
+        stdin_mock.isatty.return_value = True
+        with patch("specify_cli.events.sys") as mock_sys:
+            mock_sys.stdin = stdin_mock
+            result = _read_stdin_bounded()
+        assert result == "{}"
+
+    def test_empty_stdin_returns_empty_string(self):
+        from specify_cli.events import _read_stdin_bounded
+        from io import BytesIO
+
+        buf = BytesIO(b"")
+        stdin_mock = MagicMock()
+        stdin_mock.isatty.return_value = False
+        stdin_mock.buffer = buf
+        with patch("specify_cli.events.sys") as mock_sys:
+            mock_sys.stdin = stdin_mock
+            result = _read_stdin_bounded()
+        assert result == ""
+
+    def test_invalid_utf8_replaced(self):
+        from specify_cli.events import _read_stdin_bounded
+        from io import BytesIO
+
+        buf = BytesIO(b"hello\xff\xfeworld")
+        stdin_mock = MagicMock()
+        stdin_mock.isatty.return_value = False
+        stdin_mock.buffer = buf
+        with patch("specify_cli.events.sys") as mock_sys:
+            mock_sys.stdin = stdin_mock
+            result = _read_stdin_bounded()
+        assert "hello" in result
+        assert "world" in result
+
+
+class TestReadStdinBoundedCLI:
+    """Test the CLI event runner's bounded stdin reader."""
+
+    def test_cli_reader_small_payload(self):
+        from specify_cli.commands.event import _read_stdin_bounded
+        from io import BytesIO
+
+        buf = BytesIO(b'{"event": "test"}')
+        stdin_mock = MagicMock()
+        stdin_mock.isatty.return_value = False
+        stdin_mock.buffer = buf
+        with patch("specify_cli.commands.event.sys") as mock_sys:
+            mock_sys.stdin = stdin_mock
+            result = _read_stdin_bounded(max_bytes=1024)
+        assert result == '{"event": "test"}'
+
+    def test_cli_reader_oversized(self):
+        from specify_cli.commands.event import _read_stdin_bounded
+        from io import BytesIO
+
+        buf = BytesIO(b"y" * 500)
+        stdin_mock = MagicMock()
+        stdin_mock.isatty.return_value = False
+        stdin_mock.buffer = buf
+        with patch("specify_cli.commands.event.sys") as mock_sys:
+            mock_sys.stdin = stdin_mock
+            result = _read_stdin_bounded(max_bytes=100)
+        assert len(result) == 100
+
+    def test_cli_reader_multibyte(self):
+        from specify_cli.commands.event import _read_stdin_bounded
+        from io import BytesIO
+
+        # U+00E9 (é) is 2 bytes in UTF-8
+        payload = "caf\u00e9".encode("utf-8")
+        assert len(payload) == 5  # c + a + f + 2
+
+        buf = BytesIO(payload)
+        stdin_mock = MagicMock()
+        stdin_mock.isatty.return_value = False
+        stdin_mock.buffer = buf
+        with patch("specify_cli.commands.event.sys") as mock_sys:
+            mock_sys.stdin = stdin_mock
+            result = _read_stdin_bounded(max_bytes=5)
+        assert result == "caf\u00e9"