diff --git a/src/specify_cli/workflows/_commands.py b/src/specify_cli/workflows/_commands.py index 973e69fe0..334c7e254 100644 --- a/src/specify_cli/workflows/_commands.py +++ b/src/specify_cli/workflows/_commands.py @@ -352,6 +352,7 @@ def workflow_run( from .catalog import WorkflowRegistry registered_id: str | None = None + registry_root = project_root if not is_file_source: # Reject path-equivalent spellings ("align-wf/", "align-wf/.") that # would miss the registry lookup yet still load the installed file, @@ -364,16 +365,18 @@ 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 ID so disabled state is enforced there too. - workflows_root = (project_root / ".specify" / "workflows").resolve() + # file; map it back to its owning project and ID from the canonical + # path itself so the guard is independent of the caller's cwd. resolved = source_path.resolve() - if resolved.is_relative_to(workflows_root): - rel_parts = resolved.relative_to(workflows_root).parts - if rel_parts: - registered_id = rel_parts[0] + parts = resolved.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(resolved.anchor or ".") + registered_id = parts[i + 2] + break if registered_id is not None: - installed_meta = WorkflowRegistry(project_root).get(registered_id) + installed_meta = WorkflowRegistry(registry_root).get(registered_id) if isinstance(installed_meta, dict) and not installed_meta.get("enabled", True): err.print( f"[red]Error:[/red] Workflow '{_escape_markup(registered_id)}' is disabled. " diff --git a/src/specify_cli/workflows/catalog.py b/src/specify_cli/workflows/catalog.py index 942a6cfd0..87fe1ca0b 100644 --- a/src/specify_cli/workflows/catalog.py +++ b/src/specify_cli/workflows/catalog.py @@ -13,6 +13,7 @@ from __future__ import annotations import hashlib import json import os +import tempfile import time from dataclasses import dataclass from pathlib import Path @@ -85,15 +86,21 @@ class WorkflowRegistry: def save(self) -> None: """Persist registry to disk atomically.""" self.workflows_dir.mkdir(parents=True, exist_ok=True) - # Write-then-replace so a failed dump cannot truncate the registry. - tmp_path = self.registry_path.with_name(self.registry_path.name + ".tmp") + # Unique, exclusive temp then replace: a failed dump cannot truncate + # the registry, a pre-created symlink cannot redirect the write, and + # concurrent CLI processes cannot collide on the same temp path. + fd, tmp = tempfile.mkstemp( + dir=str(self.registry_path.parent), + prefix=f".{self.registry_path.name}.", + suffix=".tmp", + ) try: - with open(tmp_path, "w", encoding="utf-8") as f: + with os.fdopen(fd, "w", encoding="utf-8") as f: json.dump(self.data, f, indent=2) - os.replace(tmp_path, self.registry_path) - except OSError: + os.replace(tmp, self.registry_path) + except BaseException: try: - tmp_path.unlink() + os.unlink(tmp) except OSError: pass raise diff --git a/tests/test_workflows.py b/tests/test_workflows.py index 77589fb61..810a94dc3 100644 --- a/tests/test_workflows.py +++ b/tests/test_workflows.py @@ -8211,6 +8211,16 @@ steps: assert result.exit_code != 0 assert "disabled" in result.output + # Same guard must hold when invoked from outside the project. + outside = project_dir.parent / "outside-cwd" + outside.mkdir(exist_ok=True) + monkeypatch.chdir(outside) + result = runner.invoke( + app, ["workflow", "run", str(project_dir / installed_yaml)] + ) + assert result.exit_code != 0 + assert "disabled" in result.output + def test_disable_shows_marker_in_list(self, project_dir, monkeypatch): from typer.testing import CliRunner from specify_cli import app