fix(events): cap stdin in the generated dispatcher, not just the CLI command

src/specify_cli/events.py

@@ -280,11 +280,18 @@ def _run_inline(command_name, payload, project_root, timeout, envelope="plain",
     if not argv:
         return 0
     try:
+        # ``payload`` is decoded above from the binary buffer with an explicit
+        # ``utf-8``. Without ``encoding=`` here, ``text=True`` re-encodes it
+        # for the child's stdin using ``locale.getpreferredencoding()`` — on
+        # Windows that is commonly the ANSI codepage, not UTF-8, so a non-ASCII
+        # payload (e.g. ``é``) would reach the handler as the wrong bytes.
+        # Pin both directions to utf-8 so decode and re-encode agree.
         result = subprocess.run(
             argv,
             input=payload,
             capture_output=True,
             text=True,
+            encoding="utf-8",
             timeout=timeout,
             cwd=str(project_root),
         )
@@ -375,7 +382,21 @@ 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 "{}"
+    # Cap piped stdin at 1 MiB to prevent a DoS (mirrors the same guard on the
+    # `specify event run` CLI command). Read from the binary buffer so the cap
+    # counts encoded bytes, not decoded characters.
+    MAX_STDIN_BYTES = 1 * 1024 * 1024
+    if not sys.stdin.isatty():
+        raw = sys.stdin.buffer.read(MAX_STDIN_BYTES + 1)
+        if len(raw) > MAX_STDIN_BYTES:
+            print(
+                "stdin payload exceeds 1 MiB limit; truncate or pipe a smaller payload",
+                file=sys.stderr,
+            )
+            sys.exit(1)
+        payload = raw.decode("utf-8")
+    else:
+        payload = "{}"
     project_root = Path(__file__).parent.parent.resolve()
 
     # Preferred path: specify_cli is importable (durable install) — delegate to
@@ -759,11 +780,19 @@ def resolve_and_run_event_command(
         logger.warning("No script found for event command '%s'", command_name)
         return 0
     try:
+        # ``payload`` reaches here already decoded from stdin's binary buffer
+        # with an explicit ``utf-8`` (see event_run/dispatcher main()). Without
+        # ``encoding=`` here, ``text=True`` re-encodes it for the child's
+        # stdin using ``locale.getpreferredencoding()`` — on Windows that is
+        # commonly the ANSI codepage, not UTF-8, so a non-ASCII payload (e.g.
+        # ``é``) would reach the handler as the wrong bytes. Pin both
+        # directions to utf-8 so decode and re-encode agree.
         result = subprocess.run(
             argv,
             input=payload,
             capture_output=True,
             text=True,
+            encoding="utf-8",
             timeout=timeout,
             cwd=str(project_root),
         )

tests/integrations/test_events.py

@@ -1550,6 +1550,198 @@ def test_dispatcher_ignores_stale_specify_cli_without_confinement(self, tmp_path
         )
         assert not ran.exists(), f"stale package ran; stderr={result.stderr!r}"
 
+    def test_dispatcher_rejects_oversized_stdin(self, tmp_path):
+        """The generated dispatcher — the actual script native hooks invoke —
+        must enforce the same 1 MiB stdin cap as `specify event run`. The
+        #3857 DoS guard previously only applied to the CLI command; the
+        template's own `sys.stdin.read()` had no cap at all."""
+        import subprocess as _sp
+        import sys as _sys
+
+        integration = ClaudeIntegration()
+        manifest = MagicMock(spec=IntegrationManifest)
+        manifest.files = {}
+        manifest.record_file = MagicMock()
+        manifest.record_existing = MagicMock()
+        install_integration_events(
+            integration, tmp_path, manifest,
+            {"session_start": [{"command": "speckit.boot"}]},
+        )
+        dispatcher = tmp_path / EVENTS_DISPATCHER_REL
+
+        oversized = "x" * (1 * 1024 * 1024 + 10)
+        result = _sp.run(
+            [_sys.executable, str(dispatcher), "speckit.boot", "session_start", "60"],
+            input=oversized,
+            capture_output=True,
+            text=True,
+            encoding="utf-8",
+            cwd=str(tmp_path),
+        )
+        assert result.returncode == 1, f"stdout={result.stdout!r} stderr={result.stderr!r}"
+        assert "1 MiB limit" in result.stderr
+
+    def test_dispatcher_stdin_cap_counts_bytes_not_characters(self, tmp_path):
+        """~300k emoji is ~1.14 MiB of UTF-8 but only 300k *characters* —
+        comfortably under a text-mode `sys.stdin.read(N)` character cap. The
+        dispatcher must still reject it by reading from the binary buffer."""
+        import subprocess as _sp
+        import sys as _sys
+
+        integration = ClaudeIntegration()
+        manifest = MagicMock(spec=IntegrationManifest)
+        manifest.files = {}
+        manifest.record_file = MagicMock()
+        manifest.record_existing = MagicMock()
+        install_integration_events(
+            integration, tmp_path, manifest,
+            {"session_start": [{"command": "speckit.boot"}]},
+        )
+        dispatcher = tmp_path / EVENTS_DISPATCHER_REL
+
+        oversized = "\U0001F600" * 300_000  # 4 bytes each in UTF-8
+        assert len(oversized) < 1 * 1024 * 1024  # under a character-based cap
+        result = _sp.run(
+            [_sys.executable, str(dispatcher), "speckit.boot", "session_start", "60"],
+            input=oversized,
+            capture_output=True,
+            text=True,
+            encoding="utf-8",
+            cwd=str(tmp_path),
+        )
+        assert result.returncode == 1, f"stdout={result.stdout!r} stderr={result.stderr!r}"
+        assert "1 MiB limit" in result.stderr
+
+    def test_dispatcher_underlimit_stdin_still_runs(self, tmp_path):
+        """A normal, under-the-cap piped payload must still reach the handler
+        byte-for-byte, including non-ASCII content -- verifying the explicit
+        utf-8 decode of stdin and the explicit utf-8 encode of the handler's
+        subprocess input agree end-to-end (regression guard against both an
+        over-eager cap check and a locale-dependent re-encode: without an
+        explicit ``encoding=`` on the inner subprocess.run, ``text=True``
+        falls back to ``locale.getpreferredencoding()`` for the child's
+        stdin, which on Windows is commonly not UTF-8, corrupting non-ASCII
+        payloads even though the dispatcher's own stdin decode is UTF-8).
+
+        Uses a Python handler (unlike the previous POSIX-shell-only version)
+        so this test actually runs on Windows, where that mismatch occurs.
+        """
+        import subprocess as _sp
+        import sys as _sys
+
+        integration = ClaudeIntegration()
+        manifest = MagicMock(spec=IntegrationManifest)
+        manifest.files = {}
+        manifest.record_file = MagicMock()
+        manifest.record_existing = MagicMock()
+        install_integration_events(
+            integration, tmp_path, manifest,
+            {"session_start": [{"command": "speckit.boot"}]},
+        )
+        dispatcher = tmp_path / EVENTS_DISPATCHER_REL
+
+        cmd_dir = tmp_path / ".specify" / "templates" / "commands"
+        cmd_dir.mkdir(parents=True)
+        out_file = tmp_path / "payload.out"
+        (cmd_dir / "boot.md").write_text(
+            "---\ndescription: \"Boot\"\nscripts:\n  py: scripts/boot.py\n---\nBody\n",
+            encoding="utf-8",
+        )
+        script_dir = tmp_path / ".specify" / "scripts"
+        script_dir.mkdir(parents=True)
+        script = script_dir / "boot.py"
+        # Read the handler's own stdin as raw bytes (not text mode) so this
+        # script's own decoding can't mask a mismatch introduced upstream.
+        script.write_text(
+            "import sys\n"
+            f"open({str(out_file)!r}, 'wb').write(sys.stdin.buffer.read())\n",
+            encoding="utf-8",
+        )
+
+        payload = '{"key": "café"}'  # non-ASCII exercises the utf-8 round trip
+        result = _sp.run(
+            [_sys.executable, str(dispatcher), "speckit.boot", "session_start", "60"],
+            input=payload,
+            capture_output=True,
+            text=True,
+            encoding="utf-8",
+            cwd=str(tmp_path),
+        )
+        assert result.returncode == 0, f"stdout={result.stdout!r} stderr={result.stderr!r}"
+        assert out_file.read_bytes() == payload.encode("utf-8")
+
+    def test_dispatcher_inline_fallback_preserves_non_ascii_payload(self, tmp_path):
+        """The self-contained stdlib fallback (``_run_inline`` — used when
+        ``specify_cli`` is not importable, e.g. a one-time ``uvx`` init) must
+        preserve a non-ASCII payload byte-for-byte too, not just the
+        preferred delegated path.
+
+        In a dev environment where ``specify_cli`` IS importable, the
+        dispatcher always delegates to the installed
+        ``resolve_and_run_event_command`` and ``_run_inline`` is never
+        reached, so a bug isolated to ``_run_inline`` alone would not be
+        caught by ``test_dispatcher_underlimit_stdin_still_runs``. This test
+        forces the fallback the same way
+        ``test_dispatcher_ignores_stale_specify_cli_without_confinement``
+        does: a stale shadow package on PYTHONPATH lacking
+        EVENT_SCRIPT_PATH_CONFINEMENT, so the confinement check ImportErrors
+        out of delegation.
+        """
+        import subprocess as _sp
+        import sys as _sys
+
+        integration = ClaudeIntegration()
+        manifest = MagicMock(spec=IntegrationManifest)
+        manifest.files = {}
+        manifest.record_file = MagicMock()
+        manifest.record_existing = MagicMock()
+        install_integration_events(
+            integration, tmp_path, manifest,
+            {"session_start": [{"command": "speckit.boot"}]},
+        )
+        dispatcher = tmp_path / EVENTS_DISPATCHER_REL
+
+        cmd_dir = tmp_path / ".specify" / "templates" / "commands"
+        cmd_dir.mkdir(parents=True)
+        out_file = tmp_path / "payload.out"
+        (cmd_dir / "boot.md").write_text(
+            "---\ndescription: \"Boot\"\nscripts:\n  py: scripts/boot.py\n---\nBody\n",
+            encoding="utf-8",
+        )
+        script_dir = tmp_path / ".specify" / "scripts"
+        script_dir.mkdir(parents=True)
+        script = script_dir / "boot.py"
+        script.write_text(
+            "import sys\n"
+            f"open({str(out_file)!r}, 'wb').write(sys.stdin.buffer.read())\n",
+            encoding="utf-8",
+        )
+
+        fake_dir = tmp_path / "_stale_pkg"
+        pkg = fake_dir / "specify_cli"
+        pkg.mkdir(parents=True)
+        (pkg / "__init__.py").write_text("", encoding="utf-8")
+        (pkg / "events.py").write_text(
+            "def resolve_and_run_event_command(*_a, **_k):\n"
+            "    raise AssertionError('delegated path must not run')\n",
+            encoding="utf-8",
+        )
+        env = dict(os.environ)
+        env["PYTHONPATH"] = str(fake_dir)
+
+        payload = '{"key": "café"}'  # non-ASCII exercises the utf-8 round trip
+        result = _sp.run(
+            [_sys.executable, str(dispatcher), "speckit.boot", "session_start", "60"],
+            input=payload,
+            capture_output=True,
+            text=True,
+            encoding="utf-8",
+            env=env,
+            cwd=str(tmp_path),
+        )
+        assert result.returncode == 0, f"stdout={result.stdout!r} stderr={result.stderr!r}"
+        assert out_file.read_bytes() == payload.encode("utf-8")
+
     def test_dispatcher_threads_per_handler_timeout(self, tmp_path):
         """S4: the generated dispatcher reads an optional 4th timeout arg and
         uses it for the inner subprocess, instead of a fixed 120s cap that