From 0f4f3d53477717927f3de220e6158b5df39fccb9 Mon Sep 17 00:00:00 2001 From: Quratulain-bilal Date: Fri, 31 Jul 2026 16:42:14 +0500 Subject: [PATCH] fix: bound sys.stdin.read() to prevent OOM on oversized payloads sys.stdin.read() loads all of stdin into memory with no size limit. A malfunctioning agent hook can pipe gigabytes of data, exhausting process memory. Cap reads at 10 MiB in both the CLI event runner and the standalone events entry point. --- src/specify_cli/commands/event.py | 21 +++++++++++++++++++-- src/specify_cli/events.py | 19 ++++++++++++++++++- 2 files changed, 37 insertions(+), 3 deletions(-) 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