diff --git a/src/specify_cli/workflows/_commands.py b/src/specify_cli/workflows/_commands.py index 5824d2ad8..9310aa402 100644 --- a/src/specify_cli/workflows/_commands.py +++ b/src/specify_cli/workflows/_commands.py @@ -123,6 +123,109 @@ def _reject_unsafe_workflow_storage(project_root: Path) -> None: ) +def _scan_for_workflow_owner(parts: tuple[str, ...]) -> tuple[int, str] | None: + """Find the *nearest* (innermost) ``.specify/workflows/`` owner in + *parts*, scanning from the end of the path. + + Scanning from the end (rather than stopping at the first match from the + start) matters for a project nested beneath an unrelated outer path that + happens to reuse the same ``.specify``/``workflows`` segment names: the + first-from-start match would pick the outer directory and the wrong + workflow ID, silently missing the real (inner) owner's disabled check. + + Returns ``(i, workflow_id)`` where ``i`` is the index of the owning + ``.specify`` segment, or ``None`` if no owner segment is present. + """ + for i in range(len(parts) - 3, -1, -1): + if parts[i] == ".specify" and parts[i + 1] == "workflows": + return i, parts[i + 2] + return None + + +def _resolve_installed_workflow_ownership( + source_path: Path, err +) -> tuple[Path | None, str | None]: + """Map a direct ``workflow.yml`` *source_path* back to the installed + workflow (``registry_root``, ``registered_id``) it belongs to, if any. + + A path can point at installed storage two ways, both of which must + receive the same registry disabled-check: + + 1. Lexically: the path's own (symlink-preserving) segments literally + contain ``.specify/workflows/`` -- collapsing ``..``/``.`` but + never resolving symlinks, so a symlinked ``workflow.yml`` leaf (or + symlinked ```` directory) inside the owned tree is caught by the + inward-symlink refusal below rather than silently followed. + 2. Via an outward-pointing alias: *source_path* (or one of its parent + directories) is itself a symlink whose *resolved* target lands + inside some project's ``.specify/workflows//workflow.yml``, even + though the raw path used to invoke the command has no such segments + at all (e.g. ``/tmp/alias.yml`` -> a disabled installed workflow's + real file). Only the lexical case also runs the inward-symlink + refusal: the resolved case's segments are real by construction (an + already-fully-resolved path cannot itself contain a symlink + component), so that check would be a vacuous no-op there. + + Returns ``(None, None)`` when neither applies -- a genuinely standalone + external workflow file, which is allowed to run unchecked. + """ + lexical = Path(os.path.normpath(str(source_path.absolute()))) + parts = lexical.parts + match = _scan_for_workflow_owner(parts) + if match is not None: + i, registered_id = match + registry_root = Path(*parts[:i]) if i else Path(lexical.anchor or ".") + # The path-derived registry_root here may differ from the cwd's + # project_root already checked by _reject_unsafe_workflow_storage + # (e.g. this path points into another project entirely, or this + # project's own .specify is itself a symlink to an + # attacker-controlled tree) -- check it explicitly rather than + # trusting that cwd-scoped guard, and don't rely on + # WorkflowRegistry's own symlinked-parent handling as the safety + # signal here: it fails closed by raising OSError at construction + # time (see catalog.py's _load), but that surfaces as an opaque + # exception rather than this guard's clean, specific CLI error for + # the actual owning project root. + _reject_unsafe_dir(registry_root / ".specify", ".specify") + _reject_unsafe_dir( + registry_root / ".specify" / "workflows", ".specify/workflows" + ) + # A legitimately installed workflow's own directory tree never + # contains a symlink (workflow add/remove both refuse one at + # install time); one appearing here means the file actually loaded + # below would not be the file this ownership match is based on, so + # refuse rather than silently mismatch. + for k in range(i + 2, len(parts) + 1): + if Path(*parts[:k]).is_symlink(): + err.print( + "[red]Error:[/red] Refusing to run: " + f".specify/workflows/{_escape_markup(registered_id)} " + "contains a symlinked path component" + ) + raise typer.Exit(1) + return registry_root, registered_id + + # No lexical owner segments. source_path (or a parent directory) might + # still be a symlink whose resolved target lands inside some project's + # installed workflow storage under a completely unrelated-looking path + # -- check that too so an alias into a disabled workflow can't dodge + # the registry guard. + try: + resolved = source_path.resolve(strict=False) + except OSError: + return None, None + if resolved == lexical: + # Nothing on this path is a symlink; already covered above. + return None, None + resolved_parts = resolved.parts + resolved_match = _scan_for_workflow_owner(resolved_parts) + if resolved_match is None: + return None, None + i, registered_id = resolved_match + registry_root = Path(*resolved_parts[:i]) if i else Path(resolved.anchor or ".") + return registry_root, registered_id + + _WORKFLOW_ID_PATTERN = re.compile(r"^[a-z0-9](?:[a-z0-9-]*[a-z0-9])?$") _RESERVED_WORKFLOW_IDS: frozenset[str] = frozenset({"runs", "steps"}) @@ -580,48 +683,13 @@ def workflow_run( registered_id = source else: # A direct YAML path may still point at an installed workflow's own - # file; map it back to its owning project and ID from the *lexical* - # path (collapsing .. / . without resolving symlinks) rather than - # resolve(): resolving first would follow a symlinked workflow.yml - # out of .specify/workflows, fail to find an owner, and let - # engine.load_workflow below run the symlink target unchecked -- - # silently bypassing a disabled workflow's guard. - lexical = Path(os.path.normpath(str(source_path.absolute()))) - parts = lexical.parts - for i in range(len(parts) - 2): - if parts[i] == ".specify" and parts[i + 1] == "workflows": - registry_root = Path(*parts[:i]) if i else Path(lexical.anchor or ".") - registered_id = parts[i + 2] - # The path-derived registry_root here may differ from the - # cwd's project_root already checked by - # _reject_unsafe_workflow_storage above (e.g. this path - # points into another project entirely, or this project's - # own .specify is itself a symlink to an attacker-controlled - # tree) -- check it explicitly rather than trusting that - # cwd-scoped guard, and don't rely on WorkflowRegistry's own - # symlinked-parent handling below as the safety signal here: - # it now fails closed by raising OSError at construction - # time (see catalog.py's _load), but that surfaces as an - # opaque exception rather than this guard's clean, specific - # CLI error for the actual owning project root. - _reject_unsafe_dir(registry_root / ".specify", ".specify") - _reject_unsafe_dir( - registry_root / ".specify" / "workflows", ".specify/workflows" - ) - # A legitimately installed workflow's own directory tree - # never contains a symlink (workflow add/remove both refuse - # one at install time); one appearing here means the file - # actually loaded below would not be the file this ownership - # match is based on, so refuse rather than silently mismatch. - for k in range(i + 2, len(parts) + 1): - if Path(*parts[:k]).is_symlink(): - err.print( - "[red]Error:[/red] Refusing to run: " - f".specify/workflows/{_escape_markup(registered_id)} " - "contains a symlinked path component" - ) - raise typer.Exit(1) - break + # file (lexically, or via a symlinked alias pointing into installed + # storage); map it back to its owning project and ID so the + # disabled check below can't be silently bypassed. + owner_root, owner_id = _resolve_installed_workflow_ownership(source_path, err) + if owner_id is not None: + registry_root = owner_root + registered_id = owner_id if registered_id is not None: installed_meta = _open_workflow_registry(registry_root, err).get(registered_id) @@ -658,7 +726,12 @@ def workflow_run( try: with _stdout_to_stderr_when(json_output): - state = engine.execute(definition, inputs) + state = engine.execute( + definition, + inputs, + installed_workflow_id=registered_id, + installed_registry_root=registry_root if registered_id else None, + ) except ValueError as exc: err.print(f"[red]Error:[/red] {exc}") raise typer.Exit(1) @@ -700,7 +773,7 @@ def workflow_resume( ): """Resume a paused or failed workflow run.""" from . import load_custom_steps - from .engine import WorkflowEngine + from .engine import RunState, WorkflowEngine project_root = _require_specify_project() load_custom_steps(project_root) @@ -711,6 +784,38 @@ def workflow_resume( inputs = _parse_input_values(input_values, json_output=json_output) err = _error_console(json_output) + # Pre-load the persisted run state so a run started from an installed + # workflow that has since been disabled cannot resume unchecked -- + # engine.resume() replays the run directly from disk with no registry + # awareness at all, which would otherwise bypass the same disabled + # guard `workflow run` enforces. Runs without installed_workflow_id + # (a direct/non-installed source, or a run persisted before this field + # existed) are unaffected and resume exactly as before. + try: + pre_state = RunState.load(run_id, project_root) + except FileNotFoundError: + err.print(f"[red]Error:[/red] Run not found: {run_id}") + raise typer.Exit(1) + except ValueError as exc: + err.print(f"[red]Error:[/red] {exc}") + raise typer.Exit(1) + + if pre_state.installed_workflow_id is not None: + owner_root = ( + Path(pre_state.installed_registry_root) + if pre_state.installed_registry_root + else project_root + ) + installed_meta = _open_workflow_registry(owner_root, err).get( + pre_state.installed_workflow_id + ) + if isinstance(installed_meta, dict) and not installed_meta.get("enabled", True): + err.print( + f"[red]Error:[/red] Workflow '{_escape_markup(pre_state.installed_workflow_id)}' is disabled. " + f"Enable with: specify workflow enable {_escape_markup(pre_state.installed_workflow_id)}" + ) + raise typer.Exit(1) + try: with _stdout_to_stderr_when(json_output): state = engine.resume(run_id, inputs or None) diff --git a/src/specify_cli/workflows/engine.py b/src/specify_cli/workflows/engine.py index 6025c30c5..da3e2b483 100644 --- a/src/specify_cli/workflows/engine.py +++ b/src/specify_cli/workflows/engine.py @@ -428,6 +428,8 @@ class RunState: run_id: str | None = None, workflow_id: str = "", project_root: Path | None = None, + installed_workflow_id: str | None = None, + installed_registry_root: str | None = None, ) -> None: # ``run_id is None`` (omitted) → auto-generate. An explicit empty # string is *not* the same as "omitted" and must be validated like @@ -441,6 +443,15 @@ class RunState: self._validate_run_id(self.run_id) self.workflow_id = workflow_id self.project_root = project_root or Path(".") + # Identifies the installed workflow (if any) this run was started + # from, and the project root that owns its registry — set by + # execute() when the source was resolved to an installed ID (see + # workflow_run's ownership mapping). None for a direct/non-installed + # YAML source, and for any run persisted before this field existed + # (load() defaults it to None), preserving old runs' existing + # resume behavior unchanged. + self.installed_workflow_id = installed_workflow_id + self.installed_registry_root = installed_registry_root self.status = RunStatus.CREATED self.current_step_index = 0 self.current_step_id: str | None = None @@ -503,6 +514,8 @@ class RunState: state_data = { "run_id": self.run_id, "workflow_id": self.workflow_id, + "installed_workflow_id": self.installed_workflow_id, + "installed_registry_root": self.installed_registry_root, "status": self.status.value, "current_step_index": self.current_step_index, "current_step_id": self.current_step_id, @@ -559,6 +572,8 @@ class RunState: run_id=state_data["run_id"], workflow_id=state_data["workflow_id"], project_root=project_root, + installed_workflow_id=state_data.get("installed_workflow_id"), + installed_registry_root=state_data.get("installed_registry_root"), ) state.status = RunStatus(state_data["status"]) state.current_step_index = state_data.get("current_step_index", 0) @@ -654,6 +669,8 @@ class WorkflowEngine: definition: WorkflowDefinition, inputs: dict[str, Any] | None = None, run_id: str | None = None, + installed_workflow_id: str | None = None, + installed_registry_root: Path | None = None, ) -> RunState: """Execute a workflow definition. @@ -665,6 +682,12 @@ class WorkflowEngine: User-provided input values. run_id: Optional run ID (uses SPECKIT_WORKFLOW_RUN_ID when set, otherwise auto-generated). + installed_workflow_id, installed_registry_root: + When the run was started from an installed workflow (as opposed + to a direct/non-installed YAML source), identifies it and its + owning registry root so a later ``resume`` can re-check the + registry's current disabled state before continuing — see + ``workflow_resume``. Returns ------- @@ -682,6 +705,12 @@ class WorkflowEngine: run_id=effective_run_id, workflow_id=definition.id, project_root=self.project_root, + installed_workflow_id=installed_workflow_id, + installed_registry_root=( + str(installed_registry_root) + if installed_registry_root is not None + else None + ), ) # Persist a copy of the workflow definition so resume can diff --git a/tests/test_workflows.py b/tests/test_workflows.py index 4e7a8a5f7..990476223 100644 --- a/tests/test_workflows.py +++ b/tests/test_workflows.py @@ -7805,6 +7805,46 @@ steps: assert result.exit_code == 0, result.output assert WorkflowRegistry(project_dir).is_installed("align-wf") + def test_add_from_url_temp_cleanup_failure_after_success_still_exits_zero( + self, project_dir, monkeypatch + ): + """An OSError while deleting the --from download's temp file after + _validate_and_install_local() has already committed the file and + registry entry must not surface as an unhandled failure for an + install that already succeeded -- it must be a warning, exit 0.""" + from unittest.mock import patch + from typer.testing import CliRunner + from specify_cli import app + from specify_cli.workflows.catalog import WorkflowRegistry + + monkeypatch.chdir(project_dir) + data = self.WORKFLOW_YAML.format(version="1.0.0").encode() + runner = CliRunner() + + import tempfile + + real_unlink = Path.unlink + + def unlink_boom(self_path, *args, **kwargs): + if self_path.suffix == ".yml" and self_path.parent == Path(tempfile.gettempdir()): + raise OSError("permission denied") + return real_unlink(self_path, *args, **kwargs) + + with patch( + "specify_cli.authentication.http.open_url", + side_effect=lambda url, timeout=None, extra_headers=None, redirect_validator=None: self._FakeResponse(data, url), + ), pytest.MonkeyPatch.context() as mp: + mp.setattr(Path, "unlink", unlink_boom) + result = runner.invoke( + app, + ["workflow", "add", "align-wf", "--from", "https://example.com/workflow.yml"], + ) + + assert result.exit_code == 0, result.output + assert "Warning" in result.output + assert "permissiondenied" in "".join(result.output.split()) + assert WorkflowRegistry(project_dir).is_installed("align-wf") + def test_add_from_url_id_mismatch_errors(self, project_dir, monkeypatch): from unittest.mock import patch from typer.testing import CliRunner @@ -8042,6 +8082,7 @@ steps: WorkflowRegistry(project_dir) assert not (outside / "workflows").exists() + @pytest.mark.skipif(sys.platform == "win32", reason="chmod mode bits not reliable on Windows") def test_registry_save_preserves_existing_file_mode(self, project_dir): """A registry shared as 0640/0644 must keep that mode after a save, not be silently replaced by mkstemp's 0600 default -- otherwise @@ -8058,6 +8099,7 @@ steps: mode = stat.S_IMODE(registry.registry_path.stat().st_mode) assert mode == 0o644, f"expected 0644, got {oct(mode)}" + @pytest.mark.skipif(sys.platform == "win32", reason="chmod mode bits not reliable on Windows") def test_registry_save_on_new_registry_uses_secure_default_mode(self, project_dir): """A brand-new registry file (no prior mode to preserve) should keep mkstemp's secure 0600 default rather than something more permissive.""" @@ -8933,6 +8975,7 @@ steps: assert seen["validator"] is _reject_insecure_download_redirect + @pytest.mark.skipif(sys.platform == "win32", reason="chmod mode bits not reliable on Windows") def test_registry_save_failure_preserves_file_on_disk(self, project_dir, monkeypatch): """A failed dump must not truncate the persisted registry, and must not alter its on-disk mode either -- the chmod-to-match-existing-mode @@ -9463,55 +9506,96 @@ steps: assert result.exit_code != 0 assert "Failed to update" in result.output - def test_update_survives_oserror_from_backup_read(self, project_dir, monkeypatch): - """OSError while reading the backup for one workflow must not abort the whole update.""" - import json - from pathlib import Path + def test_update_registry_save_failure_restores_prior_file_without_redundant_write( + self, project_dir, monkeypatch + ): + """A registry.add() save failure during `workflow update` must be + fully restored by _install_workflow_from_catalog's own atomic + rollback (rename-based, not a byte-level rewrite). The outer + workflow_update loop must not perform any redundant write of its + own onto the destination file -- that write happened only after + typer.Exit already unwound, could itself fail/truncate the safely + preserved file, and is provably unnecessary here since the inner + transaction already restored it via rename.""" from typer.testing import CliRunner from specify_cli import app - from specify_cli.workflows.catalog import WorkflowRegistry, WorkflowCatalog + from specify_cli.workflows.catalog import WorkflowCatalog, WorkflowRegistry monkeypatch.chdir(project_dir) - registry_path = WorkflowRegistry(project_dir).registry_path - registry_path.parent.mkdir(parents=True, exist_ok=True) - registry_path.write_text( - json.dumps( - { - "schema_version": "1.0", - "workflows": { - "wobbly": { - "name": "Wobbly", - "version": "0.0.1", - "source": "catalog", - "url": "https://example.com/wobbly.yml", - }, - }, - } - ), - encoding="utf-8", - ) - # An existing installed file whose read_bytes will raise OSError. - wf_file = project_dir / ".specify" / "workflows" / "wobbly" / "workflow.yml" - wf_file.parent.mkdir(parents=True, exist_ok=True) - wf_file.write_bytes(b"schema_version: '1.0'\nworkflow:\n id: wobbly\n name: Wobbly\n version: 0.0.1\nsteps: []\n") monkeypatch.setattr( WorkflowCatalog, "get_workflow_info", - lambda self, wid: {"version": "9.9.9", "url": "https://example.com/wobbly.yml", "_install_allowed": True}, + lambda self, wid: { + "id": wid, + "name": "Align Workflow", + "version": "1.0.0", + "url": "https://example.com/workflow.yml", + "_install_allowed": True, + "_catalog_name": "test-catalog", + }, ) - real_read = Path.read_bytes - - def _boom(self, *args, **kwargs): - if self.name == "workflow.yml" and "wobbly" in str(self): - raise OSError("simulated permission denied") - return real_read(self, *args, **kwargs) - - monkeypatch.setattr(Path, "read_bytes", _boom) + original_data = self.WORKFLOW_YAML.format(version="1.0.0").encode() runner = CliRunner() - result = runner.invoke(app, ["workflow", "update"], input="y\n") - assert result.exit_code != 0, result.output - assert "Filesystem error" in result.output + with pytest.MonkeyPatch.context() as mp: + mp.setattr( + "specify_cli.authentication.http.open_url", + lambda url, timeout=None, extra_headers=None, redirect_validator=None: self._FakeResponse( + original_data, url + ), + ) + result = runner.invoke(app, ["workflow", "add", "align-wf"]) + assert result.exit_code == 0, result.output + + dest_file = project_dir / ".specify" / "workflows" / "align-wf" / "workflow.yml" + assert dest_file.read_bytes() == original_data + + new_data = self.WORKFLOW_YAML.format(version="2.0.0").encode() + + def boom_save(self): + raise OSError("disk full") + + dest_writes: list[bytes] = [] + real_write_bytes = Path.write_bytes + resolved_dest_file = dest_file.resolve() + + def tracking_write_bytes(self_path, data, *args, **kwargs): + if self_path.resolve() == resolved_dest_file: + dest_writes.append(data) + return real_write_bytes(self_path, data, *args, **kwargs) + + with pytest.MonkeyPatch.context() as mp: + mp.setattr( + WorkflowCatalog, + "get_workflow_info", + lambda self, wid: { + "id": wid, + "name": "Align Workflow", + "version": "2.0.0", + "url": "https://example.com/workflow.yml", + "_install_allowed": True, + "_catalog_name": "test-catalog", + }, + ) + mp.setattr( + "specify_cli.authentication.http.open_url", + lambda url, timeout=None, extra_headers=None, redirect_validator=None: self._FakeResponse( + new_data, url + ), + ) + mp.setattr(WorkflowRegistry, "save", boom_save) + mp.setattr(Path, "write_bytes", tracking_write_bytes) + result = runner.invoke(app, ["workflow", "update"], input="y\n") + + assert result.exit_code != 0 assert "Failed to update" in result.output + # No redundant/second write of the destination file was attempted -- + # the inner atomic commit/rollback (rename-based) is the only thing + # that ever touches it. + assert dest_writes == [] + assert dest_file.read_bytes() == original_data + registry = WorkflowRegistry(project_dir) + assert registry.is_installed("align-wf") + assert registry.get("align-wf")["version"] == "1.0.0" def test_enable_disable_corrupted_registry_entry_errors(self, project_dir, monkeypatch): import json