diff --git a/src/specify_cli/__init__.py b/src/specify_cli/__init__.py index f2ed932c46..35728b66b9 100644 --- a/src/specify_cli/__init__.py +++ b/src/specify_cli/__init__.py @@ -227,7 +227,7 @@ def ensure_executable_scripts(project_path: Path, tracker: StepTracker | None = with script.open("rb") as f: if f.read(2) != b"#!": continue - except Exception: + except OSError: continue st = script.stat() mode = st.st_mode diff --git a/src/specify_cli/_assets.py b/src/specify_cli/_assets.py index 31fb9708e6..80cd4cc3c9 100644 --- a/src/specify_cli/_assets.py +++ b/src/specify_cli/_assets.py @@ -105,7 +105,7 @@ def get_speckit_version() -> str: """Get current spec-kit version.""" try: return importlib.metadata.version("specify-cli") - except Exception: + except importlib.metadata.PackageNotFoundError: # Fallback: try reading from pyproject.toml try: import tomllib @@ -114,7 +114,7 @@ def get_speckit_version() -> str: with open(pyproject_path, "rb") as f: data = tomllib.load(f) return data.get("project", {}).get("version", "unknown") - except Exception: + except (OSError, tomllib.TOMLDecodeError, TypeError): # Intentionally ignore any errors while reading/parsing pyproject.toml. # If this lookup fails for any reason, we fall back to returning "unknown" below. pass diff --git a/src/specify_cli/_console.py b/src/specify_cli/_console.py index 8d1216387f..ccce6c20eb 100644 --- a/src/specify_cli/_console.py +++ b/src/specify_cli/_console.py @@ -84,7 +84,7 @@ def _maybe_refresh(self): if self._refresh_cb: try: self._refresh_cb() - except Exception: + except Exception: # nosec B110 pass def render(self): diff --git a/src/specify_cli/_utils.py b/src/specify_cli/_utils.py index 6603d65c45..807ab66831 100644 --- a/src/specify_cli/_utils.py +++ b/src/specify_cli/_utils.py @@ -6,7 +6,7 @@ import os import shutil import stat -import subprocess +import subprocess # nosec B404 import tempfile import yaml from pathlib import Path, PurePosixPath, PureWindowsPath @@ -86,10 +86,12 @@ def run_command( try: if capture: - result = subprocess.run(cmd, check=check_return, capture_output=True, text=True) + result = subprocess.run( # nosec B603 + cmd, check=check_return, capture_output=True, text=True + ) return result.stdout.strip() else: - subprocess.run(cmd, check=check_return) + subprocess.run(cmd, check=check_return) # nosec B603 return None except subprocess.CalledProcessError as e: if check_return: diff --git a/src/specify_cli/_version.py b/src/specify_cli/_version.py index e634a4f286..d737bf28e8 100644 --- a/src/specify_cli/_version.py +++ b/src/specify_cli/_version.py @@ -16,7 +16,7 @@ import re import shlex import shutil -import subprocess +import subprocess # nosec B404 import sys import urllib.error import urllib.parse @@ -407,7 +407,7 @@ def _uv_tool_list_contains_specify_cli(stdout: str) -> bool: if not line: continue first_token = line.split(None, 1)[0] - if first_token == "specify-cli": + if first_token == "specify-cli": # nosec B105 return True return False @@ -527,7 +527,7 @@ def _detect_install_method( if uv_bin is not None: consulted.append("uv tool list") try: - result = subprocess.run( + result = subprocess.run( # nosec B603 [uv_bin, "tool", "list"], capture_output=True, text=True, @@ -547,7 +547,7 @@ def _detect_install_method( if pipx_bin is not None: consulted.append("pipx list --json") try: - result = subprocess.run( + result = subprocess.run( # nosec B603 [pipx_bin, "list", "--json"], capture_output=True, text=True, @@ -819,7 +819,7 @@ def _run_installer(plan: _UpgradePlan) -> _InstallerResult: timeout = None try: - completed = subprocess.run( + completed = subprocess.run( # nosec B603 plan.installer_argv, shell=False, check=False, @@ -882,7 +882,7 @@ def _verify_upgrade(plan: _UpgradePlan) -> str | None: if specify_bin is None: return None try: - result = subprocess.run( + result = subprocess.run( # nosec B603 [specify_bin, "--version"], shell=False, check=False, @@ -1156,7 +1156,8 @@ def self_check() -> None: # Graceful-failure path (FR-008). `failure_reason` is one of the # enumerated strings produced by _fetch_latest_release_tag() — it # never contains a URL, headers, response body, or traceback. - assert failure_reason is not None + if failure_reason is None: + failure_reason = "unknown" console.print(f"Installed: {installed}") console.print(f"[yellow]Could not check latest release:[/yellow] {failure_reason}") return diff --git a/src/specify_cli/authentication/azure_devops.py b/src/specify_cli/authentication/azure_devops.py index 5d71a1957b..dab736e499 100644 --- a/src/specify_cli/authentication/azure_devops.py +++ b/src/specify_cli/authentication/azure_devops.py @@ -5,8 +5,9 @@ import base64 import json as _json import os -import subprocess +import subprocess # nosec B404 from typing import TYPE_CHECKING +from urllib.parse import quote from .base import AuthProvider @@ -15,6 +16,17 @@ # Azure DevOps resource ID for OAuth / Azure AD token acquisition. _ADO_RESOURCE_ID = "499b84ac-1321-427f-aa17-267ca6975798" +_MICROSOFT_LOGIN_HOST = "login.microsoftonline.com" + + +def _is_safe_tenant_segment(tenant_id: str) -> bool: + """Return True when *tenant_id* cannot alter the Azure token URL path.""" + tenant_id = tenant_id.strip() + if not tenant_id or tenant_id in (".", ".."): + return False + if any(ord(ch) < 32 or ord(ch) == 127 for ch in tenant_id): + return False + return quote(tenant_id, safe="") == tenant_id class AzureDevOpsAuth(AuthProvider): @@ -56,7 +68,7 @@ def resolve_token(self, entry: AuthConfigEntry) -> str | None: def _acquire_via_az_cli() -> str | None: """Run ``az account get-access-token`` and return the access token.""" try: - result = subprocess.run( # noqa: S603, S607 + result = subprocess.run( # noqa: S603, S607 # nosec B603 B607 [ "az", "account", @@ -91,10 +103,11 @@ def _acquire_via_client_credentials(entry: AuthConfigEntry) -> str | None: if not client_secret: return None - url = ( - f"https://login.microsoftonline.com/{entry.tenant_id}" - "/oauth2/v2.0/token" - ) + tenant_id = entry.tenant_id.strip() + if not _is_safe_tenant_segment(tenant_id): + return None + + url = f"https://{_MICROSOFT_LOGIN_HOST}/{tenant_id}/oauth2/v2.0/token" from urllib.parse import urlencode body = urlencode({ "grant_type": "client_credentials", @@ -109,7 +122,7 @@ def _acquire_via_client_credentials(entry: AuthConfigEntry) -> str | None: headers={"Content-Type": "application/x-www-form-urlencoded"}, ) try: - with urllib.request.urlopen(req, timeout=30) as resp: # noqa: S310 + with urllib.request.urlopen(req, timeout=30) as resp: # nosec B310 payload = _json.loads(resp.read().decode("utf-8")) token = payload.get("access_token", "").strip() return token or None diff --git a/src/specify_cli/authentication/http.py b/src/specify_cli/authentication/http.py index a2888bcce2..5912dd5cc4 100644 --- a/src/specify_cli/authentication/http.py +++ b/src/specify_cli/authentication/http.py @@ -25,6 +25,26 @@ _config_cache: list[AuthConfigEntry] | None = None # None = not yet loaded +def _validate_http_url(url: str) -> str: + """Return *url* when it is an absolute HTTP(S) URL with a host. + + Authenticated download helpers must never hand ``file:``, custom schemes, + relative paths, or malformed hostless URLs to ``urllib``. Keeping this as a + single validator makes both authenticated and unauthenticated fallbacks use + the same network-only boundary. + """ + if not isinstance(url, str): + raise ValueError("URL must be a string.") + if url != url.strip() or any(ord(ch) < 32 or ord(ch) == 127 for ch in url): + raise ValueError("URL must be an absolute http(s) URL with a hostname.") + parsed = urlparse(url) + if parsed.scheme not in ("http", "https") or not parsed.hostname: + raise ValueError("URL must be an absolute http(s) URL with a hostname.") + if parsed.username or parsed.password: + raise ValueError("URL must not include embedded credentials.") + return url + + def _load_config() -> list[AuthConfigEntry]: """Load auth config, using override if set (for testing). @@ -101,6 +121,7 @@ def build_request(url: str, extra_headers: dict[str, str] | None = None) -> urll Uses the first matching entry from ``auth.json`` whose token resolves. Returns a plain request when no entry matches or the file doesn't exist. """ + url = _validate_http_url(url) headers: dict[str, str] = {} if extra_headers: # Strip Authorization from extra_headers to prevent bypass @@ -150,6 +171,7 @@ def open_url( *redirect_validator*, when provided, is called with ``(old_url, new_url)`` before following each redirect and may raise to reject the redirect. """ + url = _validate_http_url(url) entries = find_entries_for_url(url, _load_config()) def _make_req(auth_headers: dict[str, str]) -> urllib.request.Request: @@ -185,4 +207,4 @@ def _make_req(auth_headers: dict[str, str]) -> urllib.request.Request: if redirect_validator is not None: opener = urllib.request.build_opener(_StripAuthOnRedirect((), redirect_validator)) return opener.open(req, timeout=timeout) - return urllib.request.urlopen(req, timeout=timeout) # noqa: S310 + return urllib.request.urlopen(req, timeout=timeout) # nosec B310 diff --git a/src/specify_cli/bundler/services/installer.py b/src/specify_cli/bundler/services/installer.py index 10f49ef94b..1f7156380b 100644 --- a/src/specify_cli/bundler/services/installer.py +++ b/src/specify_cli/bundler/services/installer.py @@ -206,5 +206,5 @@ def _rollback( for component in reversed(done): try: installer.remove(project_root, component) - except Exception: # noqa: BLE001 - best-effort rollback + except Exception: # noqa: BLE001 - best-effort rollback # nosec B112 continue diff --git a/src/specify_cli/integration_status.py b/src/specify_cli/integration_status.py index 050f0629d3..e9c1f2a9ed 100644 --- a/src/specify_cli/integration_status.py +++ b/src/specify_cli/integration_status.py @@ -381,7 +381,8 @@ def build_integration_status_report(project_root: Path) -> dict[str, Any]: ) return _build_report(None, [], findings, {}, None) - assert raw_state is not None + if raw_state is None: + return _build_report(None, [], findings, {}, None) raw_default_key = default_integration_key(raw_state) raw_installed_value = raw_state.get("installed_integrations") raw_installed_is_list = isinstance(raw_installed_value, list) diff --git a/src/specify_cli/integrations/base.py b/src/specify_cli/integrations/base.py index d5ebce78e2..87a004e995 100644 --- a/src/specify_cli/integrations/base.py +++ b/src/specify_cli/integrations/base.py @@ -266,7 +266,7 @@ def dispatch_command( Raises ``NotImplementedError`` if the integration does not support CLI dispatch. """ - import subprocess + import subprocess # nosec B404 prompt = self.build_command_invocation(command_name, args) # When streaming to the terminal, request text output so the @@ -299,7 +299,7 @@ def dispatch_command( # can Ctrl+C at any time. The timeout parameter is only # applied in the captured (non-streaming) branch below. try: - result = subprocess.run( + result = subprocess.run( # nosec B603 exec_args, text=True, cwd=cwd, @@ -316,7 +316,7 @@ def dispatch_command( "stderr": "", } - result = subprocess.run( + result = subprocess.run( # nosec B603 exec_args, capture_output=True, text=True, diff --git a/src/specify_cli/integrations/copilot/__init__.py b/src/specify_cli/integrations/copilot/__init__.py index 44bd47f353..082090f40e 100644 --- a/src/specify_cli/integrations/copilot/__init__.py +++ b/src/specify_cli/integrations/copilot/__init__.py @@ -213,7 +213,7 @@ def dispatch_command( In skills mode, the prompt includes the skill invocation (``/speckit-``). """ - import subprocess + import subprocess # nosec B404 stem = command_name if stem.startswith("speckit."): @@ -256,7 +256,7 @@ def dispatch_command( if stream: try: - result = subprocess.run( + result = subprocess.run( # nosec B603 cli_args, text=True, cwd=cwd, @@ -273,7 +273,7 @@ def dispatch_command( "stderr": "", } - result = subprocess.run( + result = subprocess.run( # nosec B603 cli_args, capture_output=True, text=True, diff --git a/src/specify_cli/integrations/generic/__init__.py b/src/specify_cli/integrations/generic/__init__.py index a2fd430f75..d0ea5a2465 100644 --- a/src/specify_cli/integrations/generic/__init__.py +++ b/src/specify_cli/integrations/generic/__init__.py @@ -63,7 +63,7 @@ def _resolve_commands_dir( import shlex tokens = shlex.split(raw) for i, token in enumerate(tokens): - if token == "--commands-dir" and i + 1 < len(tokens): + if token == "--commands-dir" and i + 1 < len(tokens): # nosec B105 return tokens[i + 1] if token.startswith("--commands-dir="): return token.split("=", 1)[1] diff --git a/src/specify_cli/presets/__init__.py b/src/specify_cli/presets/__init__.py index 863b6ef7dc..862db6f245 100644 --- a/src/specify_cli/presets/__init__.py +++ b/src/specify_cli/presets/__init__.py @@ -780,7 +780,7 @@ def _reconcile_composed_commands(self, command_names: List[str]) -> None: context_note=f"\n\n\n", ) registered = True - except Exception: + except Exception: # nosec B110 # Extension registration failed; fall back to # generic path-based registration below. pass @@ -905,7 +905,7 @@ def _register_command_from_path( cmd_tmpl["aliases"] = aliases break break - except Exception: + except Exception: # nosec B110 pass # best-effort alias loading self._register_for_non_skill_agents( registrar, [cmd_tmpl], source_id, cmd_path.parent @@ -1080,7 +1080,7 @@ def _reconcile_skills(self, command_names: List[str]) -> None: if integration is not None and hasattr(integration, "post_process_skill_content"): skill_content = integration.post_process_skill_content(skill_content) skill_file.write_text(skill_content, encoding="utf-8") - except Exception: + except Exception: # nosec B110 pass # best-effort override skill restoration # Register skills only for the specific commands being reconciled, diff --git a/src/specify_cli/workflows/__init__.py b/src/specify_cli/workflows/__init__.py index 8775428c59..6e0d6267ff 100644 --- a/src/specify_cli/workflows/__init__.py +++ b/src/specify_cli/workflows/__init__.py @@ -196,7 +196,7 @@ def load_custom_steps(project_root: Path) -> list[str]: k for k in _sys.modules if k.startswith(submodule_prefix) ]: _sys.modules.pop(_mod_key, None) - except Exception: # noqa: BLE001 + except Exception: # noqa: BLE001 # nosec B112 # Silently skip broken step packages at load time continue diff --git a/src/specify_cli/workflows/engine.py b/src/specify_cli/workflows/engine.py index 68f2ca6f3d..4ed11d42f7 100644 --- a/src/specify_cli/workflows/engine.py +++ b/src/specify_cli/workflows/engine.py @@ -59,9 +59,9 @@ def __init__(self, data: dict[str, Any], source_path: Path | None = None) -> Non # Advisory pre-conditions (spec-kit version / integrations a workflow # expects). Validated by ``validate_workflow`` (recognized keys only; # see ``_RECOGNIZED_REQUIRES_KEYS``) but NOT enforced at run time — they - # are not a security boundary. In particular there is no - # ``requires.permissions`` capability gate: shell steps always run with - # the user's privileges. + # are not a security boundary. In particular there is no granular + # ``requires.permissions`` capability sandbox; shell steps are guarded + # by their own explicit workflow-trust prompt/env gate. # # Holds the raw parsed value, so before ``validate_workflow`` runs it may # be a non-mapping (``None`` for a bare ``requires:``, a list for @@ -204,9 +204,10 @@ def validate_workflow(definition: WorkflowDefinition) -> list[str]: # integrations a workflow expects). Only a fixed set of keys is recognized; # reject anything else so authoring typos surface here instead of being # silently ignored at runtime. In particular ``requires.permissions`` is - # rejected explicitly: it reads like a runtime capability gate, but no such - # gate exists — a ``shell`` step always runs with the user's privileges, so - # declaring it would give a false sense of sandboxing. + # rejected explicitly: it reads like a granular runtime capability sandbox, + # but no such sandbox exists. A ``shell`` step has its own explicit trust + # gate, and declaring ``requires.permissions`` would still give a false + # sense of per-workflow permission enforcement. # # Mirror ``inputs`` validation: an omitted block defaults to ``{}`` and is # valid, but any present-but-non-mapping value — ``requires:`` (YAML null), @@ -219,9 +220,10 @@ def validate_workflow(definition: WorkflowDefinition) -> list[str]: if key == "permissions": errors.append( "'requires.permissions' is not a recognized or " - "enforced capability gate — shell steps always run " - "with the user's privileges. Remove it and gate " - "sensitive steps with a 'gate' step instead." + "enforced capability sandbox. Shell steps require " + "explicit workflow trust before execution, but " + "'requires.permissions' does not grant or constrain " + "runtime permissions." ) elif key not in _RECOGNIZED_REQUIRES_KEYS: errors.append( diff --git a/src/specify_cli/workflows/steps/prompt/__init__.py b/src/specify_cli/workflows/steps/prompt/__init__.py index 5ec99b794d..6317906c3c 100644 --- a/src/specify_cli/workflows/steps/prompt/__init__.py +++ b/src/specify_cli/workflows/steps/prompt/__init__.py @@ -128,14 +128,14 @@ def _try_dispatch( if not exec_args: return None - import subprocess + import subprocess # nosec B404 project_root = ( Path(context.project_root) if context.project_root else Path.cwd() ) try: - result = subprocess.run( + result = subprocess.run( # nosec B603 exec_args, text=True, cwd=str(project_root), diff --git a/src/specify_cli/workflows/steps/shell/__init__.py b/src/specify_cli/workflows/steps/shell/__init__.py index 2a65fca444..0b97d23750 100644 --- a/src/specify_cli/workflows/steps/shell/__init__.py +++ b/src/specify_cli/workflows/steps/shell/__init__.py @@ -3,7 +3,9 @@ from __future__ import annotations import json -import subprocess +import os +import subprocess # nosec B404 +import sys from typing import Any from specify_cli.workflows.base import StepBase, StepContext, StepResult, StepStatus @@ -17,6 +19,7 @@ class ShellStep(StepBase): """ type_key = "shell" + TRUST_ENV_VAR = "SPECKIT_TRUSTED_WORKFLOW" def execute(self, config: dict[str, Any], context: StepContext) -> StepResult: run_cmd = config.get("run", "") @@ -26,12 +29,37 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult: cwd = context.project_root or "." + trusted = self._is_trusted() + if not trusted: + output = { + "trusted": False, + "trust_env_var": self.TRUST_ENV_VAR, + "run": run_cmd, + } + if not sys.stdin.isatty(): + return StepResult( + status=StepStatus.PAUSED, + output=output, + error=( + "Shell step requires explicit trust before executing " + f"local commands. Re-run or resume with {self.TRUST_ENV_VAR}=1 " + "after reviewing the workflow command." + ), + ) + if not self._prompt_trust(run_cmd): + output["aborted"] = True + return StepResult( + status=StepStatus.FAILED, + output=output, + error=f"Shell step {config.get('id', '?')!r} rejected by user.", + ) + # NOTE: shell=True is required to support pipes, redirects, and # multi-command expressions in workflow YAML. Workflow authors # control commands; catalog-installed workflows should be reviewed # before use (see PUBLISHING.md for security guidance). try: - proc = subprocess.run( # noqa: S602 -- intentional shell=True (see NOTE above) + proc = subprocess.run( # noqa: S602 -- guarded shell feature; see NOTE. # nosec B602 run_cmd, shell=True, capture_output=True, @@ -84,6 +112,21 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult: output={"exit_code": -1, "stdout": "", "stderr": str(exc)}, ) + @classmethod + def _is_trusted(cls) -> bool: + return os.environ.get(cls.TRUST_ENV_VAR, "").strip() == "1" + + @staticmethod + def _prompt_trust(run_cmd: str) -> bool: + print("\n Shell workflow step wants to run this local command:") + print(f" {run_cmd}") + try: + raw = input(" Trust and run this command? [y/N]: ").strip().lower() + except (EOFError, KeyboardInterrupt): + print() + return False + return raw in {"y", "yes"} + def validate(self, config: dict[str, Any]) -> list[str]: errors = super().validate(config) if "run" not in config: diff --git a/tests/test_authentication.py b/tests/test_authentication.py index a89303d3d8..d129fb2a92 100644 --- a/tests/test_authentication.py +++ b/tests/test_authentication.py @@ -525,6 +525,36 @@ def test_resolve_token_azure_ad_network_error_returns_none(self, monkeypatch): side_effect=urllib.error.URLError("connection refused")): assert AzureDevOpsAuth().resolve_token(entry) is None + def test_resolve_token_azure_ad_rejects_path_shaped_tenant(self, monkeypatch): + """azure-ad tenant IDs must stay a single URL path segment.""" + from unittest.mock import patch + + monkeypatch.setenv("MY_SECRET", "secret-value") + entry = AuthConfigEntry( + hosts=("dev.azure.com",), provider="azure-devops", auth="azure-ad", + tenant_id="tid/../evil", client_id="cid", client_secret_env="MY_SECRET", + ) + + with patch("urllib.request.urlopen") as urlopen: + assert AzureDevOpsAuth().resolve_token(entry) is None + + urlopen.assert_not_called() + + def test_resolve_token_azure_ad_rejects_query_shaped_tenant(self, monkeypatch): + """azure-ad tenant IDs must not alter the token endpoint URL.""" + from unittest.mock import patch + + monkeypatch.setenv("MY_SECRET", "secret-value") + entry = AuthConfigEntry( + hosts=("dev.azure.com",), provider="azure-devops", auth="azure-ad", + tenant_id="tid?x=1", client_id="cid", client_secret_env="MY_SECRET", + ) + + with patch("urllib.request.urlopen") as urlopen: + assert AzureDevOpsAuth().resolve_token(entry) is None + + urlopen.assert_not_called() + # --------------------------------------------------------------------------- # open_url / build_request — positive tests @@ -691,6 +721,42 @@ def test_timeout_propagates(self, monkeypatch): with pytest.raises(socket.timeout): open_url("https://example.com/file") + @pytest.mark.parametrize( + "url", + [ + "file:///tmp/auth.json", + "ftp://example.com/file", + "https:///missing-host", + "/relative/path", + ], + ) + def test_build_request_rejects_non_http_urls(self, monkeypatch, url): + from specify_cli.authentication.http import build_request + + self._set_config(monkeypatch, []) + with pytest.raises(ValueError, match="URL must"): + build_request(url) + + @pytest.mark.parametrize( + "url", + [ + "file:///tmp/auth.json", + "ftp://example.com/file", + "https:///missing-host", + "/relative/path", + ], + ) + def test_open_url_rejects_non_http_urls_before_network(self, monkeypatch, url): + from unittest.mock import patch + from specify_cli.authentication.http import open_url + + self._set_config(monkeypatch, []) + with patch("specify_cli.authentication.http.urllib.request.urlopen") as urlopen: + with pytest.raises(ValueError, match="URL must"): + open_url(url) + + urlopen.assert_not_called() + # --------------------------------------------------------------------------- # _load_config caching diff --git a/tests/test_workflows.py b/tests/test_workflows.py index 2fdbf887b3..0e45eaae9a 100644 --- a/tests/test_workflows.py +++ b/tests/test_workflows.py @@ -47,6 +47,12 @@ def project_dir(temp_dir): return temp_dir +@pytest.fixture(autouse=True) +def trust_shell_workflow_steps_by_default(monkeypatch): + """Let legacy workflow tests execute shell steps unless they opt out.""" + monkeypatch.setenv("SPECKIT_TRUSTED_WORKFLOW", "1") + + @pytest.fixture def sample_workflow_yaml(): """Return a valid minimal workflow YAML string.""" @@ -1152,10 +1158,26 @@ def _python_run(tmp_path, body): script.write_text(body, encoding="utf-8") return f'"{sys.executable}" "{script}"' - def test_execute_echo(self): + def test_execute_untrusted_noninteractive_pauses(self, monkeypatch): + from specify_cli.workflows.steps.shell import ShellStep + from specify_cli.workflows.base import StepContext, StepStatus + + monkeypatch.delenv("SPECKIT_TRUSTED_WORKFLOW", raising=False) + + step = ShellStep() + ctx = StepContext() + config = {"id": "test", "run": "echo hello"} + result = step.execute(config, ctx) + assert result.status == StepStatus.PAUSED + assert result.output["trusted"] is False + assert "SPECKIT_TRUSTED_WORKFLOW=1" in (result.error or "") + + def test_execute_env_trusted_echo(self, monkeypatch): from specify_cli.workflows.steps.shell import ShellStep from specify_cli.workflows.base import StepContext, StepStatus + monkeypatch.setenv("SPECKIT_TRUSTED_WORKFLOW", "1") + step = ShellStep() ctx = StepContext() config = {"id": "test", "run": "echo hello"} @@ -1164,10 +1186,42 @@ def test_execute_echo(self): assert result.output["exit_code"] == 0 assert "hello" in result.output["stdout"] - def test_execute_failure(self): + def test_execute_invalid_trust_env_still_pauses(self, monkeypatch): from specify_cli.workflows.steps.shell import ShellStep from specify_cli.workflows.base import StepContext, StepStatus + monkeypatch.setenv("SPECKIT_TRUSTED_WORKFLOW", "true") + + step = ShellStep() + ctx = StepContext() + config = {"id": "test", "run": "echo hello"} + result = step.execute(config, ctx) + assert result.status == StepStatus.PAUSED + assert result.output["trusted"] is False + + def test_execute_interactive_approval_runs_command(self, monkeypatch): + import specify_cli.workflows.steps.shell as shell_module + from specify_cli.workflows.steps.shell import ShellStep + from specify_cli.workflows.base import StepContext, StepStatus + + monkeypatch.delenv("SPECKIT_TRUSTED_WORKFLOW", raising=False) + monkeypatch.setattr(shell_module.sys, "stdin", _StubStdin(True)) + monkeypatch.setattr("builtins.input", lambda _prompt: "yes") + + step = ShellStep() + ctx = StepContext() + config = {"id": "test", "run": "echo hello"} + result = step.execute(config, ctx) + assert result.status == StepStatus.COMPLETED + assert result.output["exit_code"] == 0 + assert "hello" in result.output["stdout"] + + def test_execute_failure(self, monkeypatch): + from specify_cli.workflows.steps.shell import ShellStep + from specify_cli.workflows.base import StepContext, StepStatus + + monkeypatch.setenv("SPECKIT_TRUSTED_WORKFLOW", "1") + step = ShellStep() ctx = StepContext() config = {"id": "test", "run": "exit 1"} @@ -1184,10 +1238,12 @@ def test_validate_missing_run(self): assert any("missing 'run'" in e for e in errors) - def test_output_format_json_exposes_data(self, tmp_path): + def test_output_format_json_exposes_data(self, tmp_path, monkeypatch): from specify_cli.workflows.steps.shell import ShellStep from specify_cli.workflows.base import StepContext, StepStatus + monkeypatch.setenv("SPECKIT_TRUSTED_WORKFLOW", "1") + step = ShellStep() ctx = StepContext(project_root=str(tmp_path)) config = { @@ -1202,10 +1258,12 @@ def test_output_format_json_exposes_data(self, tmp_path): assert result.output["data"] == {"items": [1, 2]} assert result.output["exit_code"] == 0 # raw keys still present - def test_output_format_json_invalid_stdout_fails(self, tmp_path): + def test_output_format_json_invalid_stdout_fails(self, tmp_path, monkeypatch): from specify_cli.workflows.steps.shell import ShellStep from specify_cli.workflows.base import StepContext, StepStatus + monkeypatch.setenv("SPECKIT_TRUSTED_WORKFLOW", "1") + step = ShellStep() ctx = StepContext(project_root=str(tmp_path)) config = { @@ -1217,10 +1275,12 @@ def test_output_format_json_invalid_stdout_fails(self, tmp_path): assert result.status == StepStatus.FAILED assert "output_format: json" in (result.error or "") - def test_no_output_format_keeps_raw_output_only(self, tmp_path): + def test_no_output_format_keeps_raw_output_only(self, tmp_path, monkeypatch): from specify_cli.workflows.steps.shell import ShellStep from specify_cli.workflows.base import StepContext, StepStatus + monkeypatch.setenv("SPECKIT_TRUSTED_WORKFLOW", "1") + step = ShellStep() ctx = StepContext(project_root=str(tmp_path)) config = { @@ -2642,10 +2702,10 @@ def test_requires_unknown_key_is_rejected(self): assert any("typo_key" in e and "requires" in e for e in errors) def test_requires_permissions_is_rejected_as_not_enforced(self): - """A `requires.permissions` block looks like a runtime capability gate - but no such gate exists — shell steps always run with the user's - privileges. Reject it explicitly so authors are not misled into - believing the declaration sandboxes execution. + """A `requires.permissions` block looks like a runtime capability + sandbox, but no granular permissions sandbox exists. Reject it + explicitly so authors are not misled into believing the declaration + grants or constrains workflow permissions. """ from specify_cli.workflows.engine import WorkflowDefinition, validate_workflow @@ -2663,11 +2723,10 @@ def test_requires_permissions_is_rejected_as_not_enforced(self): run: "echo hi" """) errors = validate_workflow(definition) - # Assert on specific markers from the intended message (the offending - # key and the `gate` remediation) so the test fails if the validation - # path or wording drifts, rather than passing on any error that merely - # happens to contain "permissions" and "not". - assert any("requires.permissions" in e and "gate" in e for e in errors) + # Assert on specific markers from the intended message so the test + # fails if the validation path or wording drifts, rather than passing + # on any error that merely happens to contain "permissions" and "not". + assert any("requires.permissions" in e and "explicit workflow trust" in e for e in errors) def test_requires_empty_sequence_is_rejected_as_non_mapping(self): """A non-mapping ``requires`` (e.g. an empty list) is an authoring