diff --git a/src/specify_cli/commands/event.py b/src/specify_cli/commands/event.py index d1576c2c70..7da8d2dcc3 100644 --- a/src/specify_cli/commands/event.py +++ b/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 diff --git a/src/specify_cli/events.py b/src/specify_cli/events.py index 3469115d6e..ced031ed99 100644 --- a/src/specify_cli/events.py +++ b/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 @@ -323,6 +347,24 @@ def _emit(output, envelope, native_event=""): sys.stdout.write(output) +_MAX_STDIN_BYTES = 10 * 1024 * 1024 # 10 MiB + + +def _read_stdin_bounded(max_bytes=_MAX_STDIN_BYTES): + """Read at most *max_bytes* from stdin to prevent unbounded memory use.""" + if sys.stdin.isatty(): + return "{}" + chunks = [] + 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") + + def main(): if len(sys.argv) < 3: sys.exit(0) @@ -347,7 +389,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 diff --git a/tests/integrations/test_events.py b/tests/integrations/test_events.py index 5dfc497b95..6ae0f15325 100644 --- a/tests/integrations/test_events.py +++ b/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"