Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
65 changes: 58 additions & 7 deletions src/specify_cli/events.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@
import sys
import subprocess
import platform
from pathlib import Path
from pathlib import Path, PurePosixPath, PureWindowsPath
from typing import TYPE_CHECKING, Any

import yaml
Expand All @@ -30,6 +30,11 @@

# -- Constants -------------------------------------------------------------

# Generated hook dispatchers refuse to delegate unless this name is True.
# An older installed specify_cli.events (uvx-init plus a stale global
# install) would otherwise run unconfined script tokens.
EVENT_SCRIPT_PATH_CONFINEMENT = True

EVENTS_DISPATCHER_DIR = Path(".specify")
EVENTS_DISPATCHER_FILENAME = "events.py"
# POSIX-form (forward-slash) relative path so it matches manifest keys, which
Expand Down Expand Up @@ -83,7 +88,22 @@
import shutil
import subprocess
import sys
from pathlib import Path
from pathlib import Path, PurePosixPath, PureWindowsPath


def _script_under_base(base, token, project_root):
"""Return token resolved under base, or None if it leaves the project."""
posix_path = PurePosixPath(token)
win_path = PureWindowsPath(token)
if posix_path.anchor or win_path.anchor:
Comment thread
mnriem marked this conversation as resolved.
return None
try:
root = project_root.resolve()
candidate = (base / token).resolve()
candidate.relative_to(root)
except (OSError, ValueError):
return None
return candidate


def _find_command_template(command_name, project_root):
Expand Down Expand Up @@ -228,8 +248,8 @@ def _resolve_argv(template_path, project_root, ext_id):
return None
if not tokens:
return None
script_abs = base / tokens[0]
if not script_abs.exists():
script_abs = _script_under_base(base, tokens[0], project_root)
if script_abs is None or not script_abs.exists():
return None
rest = tokens[1:]

Expand Down Expand Up @@ -361,8 +381,15 @@ def main():
# Preferred path: specify_cli is importable (durable install) — delegate to
# the full resolver, which also handles extension manifests whose file stem
# differs from the command name and the project's custom script selection.
# Require EVENT_SCRIPT_PATH_CONFINEMENT so a stale global install cannot
# bypass the generated dispatcher's path guard.
try:
from specify_cli.events import resolve_and_run_event_command
from specify_cli.events import (
EVENT_SCRIPT_PATH_CONFINEMENT as _confine_ok,
resolve_and_run_event_command,
)
if _confine_ok is not True:
raise ImportError("specify_cli.events lacks script path confinement")
sys.exit(
resolve_and_run_event_command(
command_name, _event_name, payload, project_root, timeout=timeout, envelope=envelope, native_event=native_event
Expand Down Expand Up @@ -541,6 +568,30 @@ def _find_command_template(command_name: str, project_root: Path) -> tuple[Path
return None, None


def _confine_event_script_path(
project_root: Path, base: Path, token: str
) -> Path | None:
"""Resolve *token* under *base*, or None if it leaves the project.

Rejects anchored tokens (absolute, drive, UNC) so ``Path`` cannot
discard *base*. ``..`` is allowed when the resolved path stays inside
*project_root*, which is how extension templates reach core scripts
via ``../../scripts/...``. Keep the generated ``_script_under_base``
in sync.
"""
posix_path = PurePosixPath(token)
win_path = PureWindowsPath(token)
if posix_path.anchor or win_path.anchor:
return None
try:
root = project_root.resolve()
candidate = (base / token).resolve()
candidate.relative_to(root)
except (OSError, ValueError):
return None
return candidate


def _resolve_event_command_argv(
template_path: Path, project_root: Path, ext_id: str | None
) -> list[str] | None:
Expand Down Expand Up @@ -609,8 +660,8 @@ def _resolve_event_command_argv(
return None
if not tokens:
return None
script_abs = base / tokens[0]
if not script_abs.exists():
script_abs = _confine_event_script_path(project_root, base, tokens[0])
if script_abs is None or not script_abs.exists():
return None
rest_args = tokens[1:]

Expand Down
220 changes: 218 additions & 2 deletions tests/integrations/test_events.py
Original file line number Diff line number Diff line change
Expand Up @@ -1388,8 +1388,10 @@ def test_dispatcher_is_self_contained(self, tmp_path):
{"pre_tool_use": [{"command": "speckit.tdd.validate"}]},
)
content = (tmp_path / EVENTS_DISPATCHER_REL).read_text()
# Delegates to specify_cli when importable.
assert "from specify_cli.events import resolve_and_run_event_command" in content
# Delegates to specify_cli when importable and confinement is present.
assert "EVENT_SCRIPT_PATH_CONFINEMENT" in content
assert "from specify_cli.events import" in content
assert "resolve_and_run_event_command" in content
assert "except (ImportError, TypeError):" in content
# Inline stdlib fallback resolver for one-time/temporary installs.
assert "_run_inline" in content
Expand Down Expand Up @@ -1454,6 +1456,53 @@ def test_dispatcher_inline_fallback_runs_script(self, tmp_path):
assert out_file.exists(), f"inline fallback did not run script; stderr={result.stderr!r} rc={result.returncode}"
assert out_file.read_text() == '{"tool_name":"x"}'

def test_dispatcher_ignores_stale_specify_cli_without_confinement(self, tmp_path):
"""A generated dispatcher must not delegate to an older specify_cli
that lacks EVENT_SCRIPT_PATH_CONFINEMENT (uvx-init plus stale
global install). Absolute script tokens stay rejected."""
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)
ran = tmp_path / "stale-ran"
(cmd_dir / "boot.md").write_text(
"---\ndescription: \"Boot\"\nscripts:\n sh: /tmp/outside.sh\n---\nBody\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"
f" open({str(ran)!r}, 'w').write('delegated')\n"
" return 0\n",
encoding="utf-8",
)
env = dict(os.environ)
env["PYTHONPATH"] = str(fake_dir)
result = _sp.run(
[_sys.executable, str(dispatcher), "speckit.boot", "session_start", "60"],
input="{}",
capture_output=True,
text=True,
env=env,
cwd=str(tmp_path),
)
assert not ran.exists(), f"stale package ran; stderr={result.stderr!r}"

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
Expand Down Expand Up @@ -1522,6 +1571,173 @@ def test_sh_variant_uses_launcher_on_windows(self, tmp_path):
else:
assert PurePath(argv[0]).as_posix().endswith(".specify/scripts/bash/boot.sh")

def test_absolute_script_token_returns_none(self, tmp_path):
"""An absolute first ``scripts:`` token must not run a host binary."""
from specify_cli.events import _resolve_event_command_argv

outside = tmp_path.parent / "outside-event-script.sh"
outside.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
cmd_dir = tmp_path / ".specify" / "templates" / "commands"
cmd_dir.mkdir(parents=True)
(cmd_dir / "boot.md").write_text(
"---\n"
"description: \"Boot\"\n"
f"scripts:\n sh: {outside.as_posix()}\n"
"---\nBody\n",
encoding="utf-8",
)

argv = _resolve_event_command_argv(cmd_dir / "boot.md", tmp_path, None)

assert argv is None

def test_dotdot_script_token_outside_project_returns_none(self, tmp_path):
"""A ``..`` walk out of the project root must not resolve."""
from specify_cli.events import _resolve_event_command_argv

outside = tmp_path.parent / "outside-event-script.sh"
outside.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
cmd_dir = tmp_path / ".specify" / "templates" / "commands"
cmd_dir.mkdir(parents=True)
(cmd_dir / "boot.md").write_text(
"---\n"
"description: \"Boot\"\n"
"scripts:\n sh: ../../outside-event-script.sh\n"
"---\nBody\n",
encoding="utf-8",
)

argv = _resolve_event_command_argv(cmd_dir / "boot.md", tmp_path, None)

assert argv is None

def test_extension_dotdot_to_core_scripts_resolves(self, tmp_path):
"""Extension templates may reach core scripts via ``../../scripts/...``."""
from specify_cli.events import _resolve_event_command_argv

ext_id = "my-ext"
cmd_dir = tmp_path / ".specify" / "extensions" / ext_id / "commands"
cmd_dir.mkdir(parents=True)
(cmd_dir / "boot.md").write_text(
"---\n"
"description: \"Boot\"\n"
"scripts:\n sh: ../../scripts/bash/helper.sh\n"
"---\nBody\n",
encoding="utf-8",
)
helper_dir = tmp_path / ".specify" / "scripts" / "bash"
helper_dir.mkdir(parents=True)
(helper_dir / "helper.sh").write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")

argv = _resolve_event_command_argv(cmd_dir / "boot.md", tmp_path, ext_id)

assert argv is not None
script_arg = argv[1] if platform.system().lower().startswith("win") else argv[0]
assert PurePath(script_arg).as_posix().endswith(".specify/scripts/bash/helper.sh")

def test_symlink_escape_returns_none(self, tmp_path):
"""A relative token that resolves through a symlink out of the project
must not run the host target."""
from specify_cli.events import _resolve_event_command_argv

host = tmp_path.parent / "host-event-script.sh"
host.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
script_dir = tmp_path / ".specify" / "scripts"
script_dir.mkdir(parents=True)
sneak = script_dir / "sneak.sh"
try:
sneak.symlink_to(host)
except OSError:
pytest.skip("symlinks are not available")
cmd_dir = tmp_path / ".specify" / "templates" / "commands"
cmd_dir.mkdir(parents=True)
(cmd_dir / "boot.md").write_text(
"---\n"
"description: \"Boot\"\n"
"scripts:\n sh: scripts/sneak.sh\n"
"---\nBody\n",
encoding="utf-8",
)

argv = _resolve_event_command_argv(cmd_dir / "boot.md", tmp_path, None)

assert argv is None

def test_windows_drive_script_token_returns_none(self, tmp_path):
"""A Windows-anchored first token must not discard the project base."""
from specify_cli.events import _resolve_event_command_argv

cmd_dir = tmp_path / ".specify" / "templates" / "commands"
cmd_dir.mkdir(parents=True)
(cmd_dir / "boot.md").write_text(
"---\n"
"description: \"Boot\"\n"
"scripts:\n sh: C:/Windows/System32/cmd.exe\n"
"---\nBody\n",
encoding="utf-8",
)

argv = _resolve_event_command_argv(cmd_dir / "boot.md", tmp_path, None)

assert argv is None

def test_dispatcher_template_confines_script_token(self):
"""The stdlib fallback dispatcher must carry the same confinement."""
from specify_cli.events import _EVENTS_DISPATCHER_TEMPLATE

assert "_script_under_base" in _EVENTS_DISPATCHER_TEMPLATE
assert "PureWindowsPath" in _EVENTS_DISPATCHER_TEMPLATE

def test_dispatcher_inline_rejects_absolute_script(self, tmp_path):
"""Inline fallback must not execute an absolute first ``scripts:`` token."""
import subprocess as _sp
import sys as _sys

if platform.system().lower().startswith("win"):
return

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
marker = tmp_path / "should-not-run.out"
host = tmp_path.parent / "host-boot.sh"
host.write_text(
f"#!/bin/sh\necho ran > {shlex.quote(str(marker))}\nexit 0\n",
encoding="utf-8",
)
host.chmod(0o755)
cmd_dir = tmp_path / ".specify" / "templates" / "commands"
cmd_dir.mkdir(parents=True)
(cmd_dir / "boot.md").write_text(
"---\n"
"description: \"Boot\"\n"
f"scripts:\n sh: {host.as_posix()}\n"
"---\nBody\n",
encoding="utf-8",
)
fake_dir = tmp_path / "_fake"
(fake_dir / "specify_cli").mkdir(parents=True)
(fake_dir / "specify_cli" / "__init__.py").write_text("", encoding="utf-8")
env = dict(os.environ)
env["PYTHONPATH"] = str(fake_dir)
result = _sp.run(
[_sys.executable, str(dispatcher), "speckit.boot", "session_start", "60"],
input="{}",
capture_output=True,
text=True,
env=env,
cwd=str(tmp_path),
)
assert result.returncode == 0, result.stderr
assert not marker.exists()


# -- Merge/teardown idempotency & safety (Tier 3) ----------------------------

Expand Down