diff --git a/src/specify_cli/commands/event.py b/src/specify_cli/commands/event.py index 764fde7f6b..7e50592a6e 100644 --- a/src/specify_cli/commands/event.py +++ b/src/specify_cli/commands/event.py @@ -6,6 +6,23 @@ 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.""" + if sys.stdin.isatty(): + return "{}" + chunks: list[str] = [] + total = 0 + while total < max_bytes: + chunk = sys.stdin.read(min(max_bytes - total, 65536)) + if not chunk: + break + chunks.append(chunk) + total += len(chunk) + return "".join(chunks) + event_app = typer.Typer( name="event", help="Manage and execute event-driven commands", @@ -24,8 +41,8 @@ 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 - payload = sys.stdin.read() if not sys.stdin.isatty() else "{}" + # Read payload from stdin if available (bounded to prevent OOM) + 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 98e49aee36..d0e9b2a1ff 100644 --- a/src/specify_cli/events.py +++ b/src/specify_cli/events.py @@ -20,6 +20,23 @@ from pathlib import Path from typing import TYPE_CHECKING, Any +_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.""" + if sys.stdin.isatty(): + return "{}" + chunks: list[str] = [] + total = 0 + while total < max_bytes: + chunk = sys.stdin.read(min(max_bytes - total, 65536)) + if not chunk: + break + chunks.append(chunk) + total += len(chunk) + return "".join(chunks) + import yaml if TYPE_CHECKING: @@ -297,7 +314,7 @@ def main(): timeout = int(sys.argv[3]) except (TypeError, ValueError): timeout = 120 - 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