diff --git a/src/specify_cli/workflows/command_status.py b/src/specify_cli/workflows/command_status.py index 97471eb10c..b608c6d40b 100644 --- a/src/specify_cli/workflows/command_status.py +++ b/src/specify_cli/workflows/command_status.py @@ -100,7 +100,12 @@ def workflow_status( f"{cli._escape_markup(scope['workflow_id'])}: {scope['status']}" ) else: - runs = engine.list_runs() + err = cli._error_console(json_output) + try: + runs = engine.list_runs() + except (OSError, ValueError) as exc: + err.print(f"[red]Error:[/red] {cli._escape_markup(str(exc))}") + raise cli.typer.Exit(1) if json_output: payload = { diff --git a/src/specify_cli/workflows/engine.py b/src/specify_cli/workflows/engine.py index 5e85353834..eb4f910ef2 100644 --- a/src/specify_cli/workflows/engine.py +++ b/src/specify_cli/workflows/engine.py @@ -13,9 +13,9 @@ import json import os import re -import tempfile import threading import uuid +from contextlib import contextmanager from contextvars import ContextVar from datetime import datetime, timezone from pathlib import Path @@ -610,6 +610,10 @@ def _validate_steps( class RunState: """Manages workflow run state for persistence and resume.""" + _RUN_ARTIFACTS = frozenset( + {"state.json", "inputs.json", "workflow.yml", "log.jsonl"} + ) + # ``run_id`` is interpolated into a filesystem path (``runs/``) # by both ``save()`` and ``load()``. Constrain it to a charset that # cannot contain path separators (``/`` ``\``), parent-directory @@ -637,6 +641,111 @@ def _validate_run_id(cls, run_id: str) -> None: "alphanumeric character)." ) + @classmethod + def _run_directory( + cls, project_root: Path, run_id: str, *, create: bool = False + ) -> Path: + """Return a run directory after enforcing the persisted-run boundary.""" + cls._validate_run_id(run_id) + runs_root = project_root / ".specify" / "workflows" / "runs" + if create: + runs_root.mkdir(parents=True, exist_ok=True) + cls._reject_symlink(runs_root, ".specify/workflows/runs") + + run_dir = runs_root / run_id + if run_dir.is_symlink(): + raise ValueError( + f"Refusing to use symlinked workflow run path: {run_dir}" + ) + if run_dir.exists() and not run_dir.is_dir(): + raise OSError(f"Workflow run path is not a directory: {run_dir}") + if create: + run_dir.mkdir(parents=True, exist_ok=True) + return run_dir + + @staticmethod + def _reject_symlink(path: Path, label: str) -> None: + """Reject an existing symlink without resolving its target.""" + if path.is_symlink(): + raise ValueError(f"Refusing to use symlinked {label} path: {path}") + + @staticmethod + def _supports_descriptor_access() -> bool: + supports_dir_fd = getattr(os, "supports_dir_fd", ()) + return ( + hasattr(os, "O_DIRECTORY") + and hasattr(os, "O_NOFOLLOW") + and os.open in supports_dir_fd + and os.rename in supports_dir_fd + and os.unlink in supports_dir_fd + ) + + @classmethod + def _validate_fallback_path(cls, path: Path) -> None: + """Recheck persisted-run components before path-based access.""" + cls._reject_symlink(path.parent.parent, ".specify/workflows/runs") + cls._reject_symlink(path.parent, "workflow run") + cls._reject_symlink(path, f"workflow run artifact {path.name!r}") + + @classmethod + def _run_artifact( + cls, project_root: Path, run_id: str, name: str, *, create: bool = False + ) -> Path: + if name not in cls._RUN_ARTIFACTS: + raise ValueError(f"Unknown workflow run artifact: {name}") + run_dir = cls._run_directory(project_root, run_id, create=create) + artifact = run_dir / name + cls._reject_symlink(artifact, f"workflow run artifact {name!r}") + return artifact + + @classmethod + @contextmanager + def _open_run_directory(cls, project_root: Path, run_id: str, *, create=False): + """Open a run directory without following a replacement symlink.""" + run_path = cls._run_directory(project_root, run_id, create=create) + if not cls._supports_descriptor_access(): + yield None + return + + directory_flag = os.O_DIRECTORY + nofollow = os.O_NOFOLLOW + runs_path = run_path.parent + runs_fd = os.open(runs_path, os.O_RDONLY | directory_flag | nofollow) + try: + run_fd = os.open( + run_id, + os.O_RDONLY | directory_flag | nofollow, + dir_fd=runs_fd, + ) + finally: + os.close(runs_fd) + try: + yield run_fd + finally: + os.close(run_fd) + + @staticmethod + def _open_artifact( + path: Path, mode: str, *, dir_fd: int | None = None, **kwargs + ): + """Open a run artifact relative to its safely opened run directory.""" + if dir_fd is None: + RunState._validate_fallback_path(path) + return open(path, mode, **kwargs) + + flags = { + "r": os.O_RDONLY, + "a": os.O_WRONLY | os.O_CREAT | os.O_APPEND, + "w": os.O_WRONLY | os.O_CREAT | os.O_TRUNC, + }.get(mode) + if flags is None: + raise ValueError(f"Unsupported workflow artifact open mode: {mode!r}") + nofollow = getattr(os, "O_NOFOLLOW", None) + if nofollow is None or os.open not in os.supports_dir_fd: + raise OSError("Secure workflow artifact access is unavailable") + fd = os.open(path.name, flags | nofollow, 0o666, dir_fd=dir_fd) + return os.fdopen(fd, mode, **kwargs) + @staticmethod def _validate_installed_origin( installed_workflow_id: str | None, @@ -734,7 +843,7 @@ def __init__( @property def runs_dir(self) -> Path: - return self.project_root / ".specify" / "workflows" / "runs" / self.run_id + return self._run_directory(self.project_root, self.run_id) def save(self) -> None: """Persist current state to disk. @@ -744,8 +853,7 @@ def save(self) -> None: nor leave a reader observing a half-written file. Racing writers only contend to be last; they never corrupt. """ - runs_dir = self.runs_dir - runs_dir.mkdir(parents=True, exist_ok=True) + self._run_directory(self.project_root, self.run_id, create=True) with self._lock: if self._checkpoint_failed: @@ -773,8 +881,19 @@ def save(self) -> None: if self.execution is not None: state_data["execution"] = self.execution try: - self._atomic_write_json(runs_dir / "state.json", state_data) - self._atomic_write_json(runs_dir / "inputs.json", {"inputs": self.inputs}) + state_path = self._run_artifact( + self.project_root, self.run_id, "state.json", create=True + ) + inputs_path = self._run_artifact( + self.project_root, self.run_id, "inputs.json", create=True + ) + with self._open_run_directory( + self.project_root, self.run_id, create=True + ) as run_fd: + self._atomic_write_json(state_path, state_data, dir_fd=run_fd) + self._atomic_write_json( + inputs_path, {"inputs": self.inputs}, dir_fd=run_fd + ) except Exception as exc: # Only write failures poison the instance. A graceful interrupt # (KeyboardInterrupt) propagates unchanged: during step @@ -788,10 +907,35 @@ def save(self) -> None: raise CheckpointError(str(exc)) from exc @staticmethod - def _atomic_write_json(path: Path, data: dict[str, Any]) -> None: - """Write *data* as indented JSON to *path* atomically (temp + ``os.replace``).""" - fd, tmp = tempfile.mkstemp( - dir=str(path.parent), prefix=f".{path.name}.", suffix=".tmp" + def _atomic_write_json( + path: Path, data: dict[str, Any], *, dir_fd: int | None = None + ) -> None: + """Write *data* as indented JSON to *path* atomically.""" + tmp = f".{path.name}.{uuid.uuid4().hex}.tmp" + if dir_fd is None: + tmp_path = path.with_name(tmp) + RunState._validate_fallback_path(path) + try: + with tmp_path.open( + "x", encoding="utf-8", errors="backslashreplace" + ) as f: + json.dump(data, f, indent=2, ensure_ascii=False) + RunState._validate_fallback_path(path) + os.replace(tmp_path, path) + except BaseException: + try: + RunState._validate_fallback_path(tmp_path) + tmp_path.unlink() + except (OSError, ValueError): + pass + raise + return + + fd = os.open( + tmp, + os.O_WRONLY | os.O_CREAT | os.O_EXCL, + 0o600, + dir_fd=dir_fd, ) try: # Keep non-ASCII text readable in these human-auditable records; @@ -800,10 +944,12 @@ def _atomic_write_json(path: Path, data: dict[str, Any]) -> None: fd, "w", encoding="utf-8", errors="backslashreplace" ) as f: json.dump(data, f, indent=2, ensure_ascii=False) - os.replace(tmp, path) + os.rename( + tmp, path.name, src_dir_fd=dir_fd, dst_dir_fd=dir_fd + ) except BaseException: try: - os.unlink(tmp) + os.unlink(tmp, dir_fd=dir_fd) except OSError: pass raise @@ -823,13 +969,14 @@ def load(cls, run_id: str, project_root: Path) -> RunState: *after* the file lookup — too late to prevent the disclosure. Mirrors the precedent in ``agents._ensure_within_directory``. """ - cls._validate_run_id(run_id) - runs_dir = project_root / ".specify" / "workflows" / "runs" / run_id - state_path = runs_dir / "state.json" + state_path = cls._run_artifact(project_root, run_id, "state.json") try: - with open(state_path, encoding="utf-8") as f: - state_data = json.load(f) + with cls._open_run_directory(project_root, run_id) as run_fd: + with cls._open_artifact( + state_path, "r", dir_fd=run_fd, encoding="utf-8" + ) as f: + state_data = json.load(f) except FileNotFoundError: msg = f"Run state not found: {state_path}" raise FileNotFoundError(msg) @@ -920,20 +1067,27 @@ def load(cls, run_id: str, project_root: Path) -> RunState: validate_execution(state.execution) - inputs_path = runs_dir / "inputs.json" - if "inputs" not in state_data and inputs_path.exists(): - with open(inputs_path, encoding="utf-8") as f: - inputs_data = json.load(f) - if not isinstance(inputs_data, dict): - raise ValueError( - "Invalid run inputs: expected a JSON object" - ) - inputs = inputs_data.get("inputs", {}) - if not isinstance(inputs, dict): - raise ValueError( - "Invalid run inputs: 'inputs' must be a JSON object" - ) - state.inputs = inputs + if "inputs" not in state_data: + inputs_path = cls._run_artifact(project_root, run_id, "inputs.json") + try: + with cls._open_run_directory(project_root, run_id) as run_fd: + with cls._open_artifact( + inputs_path, "r", dir_fd=run_fd, encoding="utf-8" + ) as f: + inputs_data = json.load(f) + except FileNotFoundError: + inputs_data = None + if inputs_data is not None: + if not isinstance(inputs_data, dict): + raise ValueError( + "Invalid run inputs: expected a JSON object" + ) + inputs = inputs_data.get("inputs", {}) + if not isinstance(inputs, dict): + raise ValueError( + "Invalid run inputs: 'inputs' must be a JSON object" + ) + state.inputs = inputs if "inputs" in state_data: if not isinstance(state_data["inputs"], dict): @@ -951,8 +1105,7 @@ def append_log(self, entry: dict[str, Any]) -> None: worker can log once any checkpoint of this run has failed. """ entry["timestamp"] = datetime.now(timezone.utc).isoformat() - runs_dir = self.runs_dir - runs_dir.mkdir(parents=True, exist_ok=True) + self._run_directory(self.project_root, self.run_id, create=True) with self._lock: if self._checkpoint_failed: from ._execution import CheckpointError @@ -960,13 +1113,20 @@ def append_log(self, entry: dict[str, Any]) -> None: raise CheckpointError("A previous checkpoint failed; reload the run") with self._log_lock: self.log_entries.append(entry) - with open( - runs_dir / "log.jsonl", - "a", - encoding="utf-8", - errors="backslashreplace", - ) as f: - f.write(json.dumps(entry, ensure_ascii=False) + "\n") + log_path = self._run_artifact( + self.project_root, self.run_id, "log.jsonl", create=True + ) + with self._open_run_directory( + self.project_root, self.run_id, create=True + ) as run_fd: + with self._open_artifact( + log_path, + "a", + dir_fd=run_fd, + encoding="utf-8", + errors="backslashreplace", + ) as f: + f.write(json.dumps(entry, ensure_ascii=False) + "\n") # -- Workflow Engine ------------------------------------------------------ @@ -1110,12 +1270,17 @@ def execute( # Persist a copy of the workflow definition so resume can # reload it even if the original source is no longer available # (e.g. a local YAML path that was moved or deleted). - run_dir = self.project_root / ".specify" / "workflows" / "runs" / state.run_id - run_dir.mkdir(parents=True, exist_ok=True) - workflow_copy = run_dir / "workflow.yml" + workflow_copy = RunState._run_artifact( + self.project_root, state.run_id, "workflow.yml", create=True + ) import yaml - with open(workflow_copy, "w", encoding="utf-8") as f: - yaml.safe_dump(definition.data, f, sort_keys=False, allow_unicode=True) + with RunState._open_run_directory( + self.project_root, state.run_id, create=True + ) as run_fd: + with RunState._open_artifact( + workflow_copy, "w", dir_fd=run_fd, encoding="utf-8" + ) as f: + yaml.safe_dump(definition.data, f, sort_keys=False, allow_unicode=True) # Resolve inputs resolved_inputs = self._resolve_inputs(definition, inputs or {}) @@ -1182,12 +1347,29 @@ def resume( # Load the workflow definition — try the persisted copy in the # run directory first so resume works even if the original # source (e.g. a local YAML path) is no longer available. - run_dir = self.project_root / ".specify" / "workflows" / "runs" / run_id - run_copy = run_dir / "workflow.yml" - if run_copy.exists(): - definition = WorkflowDefinition.from_yaml(run_copy) - else: + run_copy = RunState._run_artifact( + self.project_root, run_id, "workflow.yml" + ) + try: + with RunState._open_run_directory(self.project_root, run_id) as run_fd: + with RunState._open_artifact( + run_copy, "r", dir_fd=run_fd, encoding="utf-8" + ) as f: + try: + data = yaml.safe_load(f) + except yaml.YAMLError as exc: + msg = f"Invalid YAML in {run_copy}: {exc}" + raise ValueError(msg) from exc + except FileNotFoundError: definition = self.load_workflow(state.workflow_id) + else: + if not isinstance(data, dict): + msg = ( + "Workflow YAML must be a mapping, " + f"got {type(data).__name__}." + ) + raise ValueError(msg) + definition = WorkflowDefinition(data, source_path=run_copy) # RunState.load() rejects a non-int/negative current_step_index but # can't check the upper bound — the step count isn't known until the @@ -1484,21 +1666,35 @@ def list_runs(self) -> list[dict[str, Any]]: runs_dir = self.project_root / ".specify" / "workflows" / "runs" if not runs_dir.exists(): return [] + RunState._reject_symlink(runs_dir, ".specify/workflows/runs") runs: list[dict[str, Any]] = [] for run_dir in sorted(runs_dir.iterdir()): + if run_dir.is_symlink(): + raise ValueError( + f"Refusing to use symlinked workflow run path: {run_dir}" + ) if not run_dir.is_dir(): continue - state_path = run_dir / "state.json" - if state_path.exists(): - try: - with open(state_path, encoding="utf-8") as f: + state_path = RunState._run_artifact( + self.project_root, run_dir.name, "state.json" + ) + try: + with RunState._open_run_directory( + self.project_root, run_dir.name + ) as run_fd: + with RunState._open_artifact( + state_path, + "r", + dir_fd=run_fd, + encoding="utf-8", + ) as f: state_data = json.load(f) - except (json.JSONDecodeError, OSError, UnicodeDecodeError): - continue - if not isinstance(state_data, dict) or "run_id" not in state_data: - continue - runs.append(state_data) + except (json.JSONDecodeError, OSError, UnicodeDecodeError): + continue + if not isinstance(state_data, dict) or "run_id" not in state_data: + continue + runs.append(state_data) return runs diff --git a/tests/specify_cli/workflows/test_command_status.py b/tests/specify_cli/workflows/test_command_status.py index a29a4001eb..7b25c38330 100644 --- a/tests/specify_cli/workflows/test_command_status.py +++ b/tests/specify_cli/workflows/test_command_status.py @@ -3,13 +3,13 @@ from __future__ import annotations import json +import os from pathlib import Path import pytest import typer - class TestWorkflowJsonOutput: """Test the --json machine-readable output for run/resume/status.""" @@ -95,6 +95,36 @@ def test_status_json_single_and_list(self, project_dir): ) assert any(r["run_id"] == rid for r in listing["runs"]) + @pytest.mark.skipif(not hasattr(os, "symlink"), reason="symlinks are unavailable") + @pytest.mark.parametrize("symlink_target", ["run-directory", "state.json"]) + def test_status_json_list_rejects_symlinked_storage( + self, project_dir, tmp_path, symlink_target + ): + runs_dir = project_dir / ".specify" / "workflows" / "runs" + runs_dir.mkdir(parents=True) + outside = tmp_path / "outside" + outside.mkdir() + (outside / "state.json").write_text( + json.dumps({ + "run_id": "run-1", + "workflow_id": "workflow", + "status": "paused", + }), + encoding="utf-8", + ) + if symlink_target == "run-directory": + (runs_dir / "run-1").symlink_to(outside, target_is_directory=True) + else: + run_dir = runs_dir / "run-1" + run_dir.mkdir() + (run_dir / "state.json").symlink_to(outside / "state.json") + + result = self._invoke(project_dir, ["workflow", "status", "--json"]) + + assert result.exit_code == 1 + assert result.stdout == "" + assert "symlinked workflow run" in result.stderr + def test_composed_gate_status_and_resume(self, project_dir): from specify_cli.workflows.catalog import WorkflowRegistry diff --git a/tests/specify_cli/workflows/test_run_state_storage.py b/tests/specify_cli/workflows/test_run_state_storage.py new file mode 100644 index 0000000000..203b2e9156 --- /dev/null +++ b/tests/specify_cli/workflows/test_run_state_storage.py @@ -0,0 +1,283 @@ +from __future__ import annotations + +import json +import os + +import pytest + + +pytestmark = pytest.mark.skipif( + not hasattr(os, "symlink"), reason="symlinks are unavailable" +) + + +def _run_dir(project_dir, run_id="run-1"): + path = project_dir / ".specify" / "workflows" / "runs" / run_id + path.mkdir(parents=True, exist_ok=True) + return path + + +def _state(project_dir, run_id="run-1"): + from specify_cli.workflows.base import RunStatus + from specify_cli.workflows.engine import RunState + + state = RunState(run_id=run_id, workflow_id="workflow", project_root=project_dir) + state.status = RunStatus.PAUSED + return state + + +def test_workflow_operations_use_portable_storage_fallback( + project_dir, monkeypatch +): + from specify_cli.workflows.base import RunStatus + from specify_cli.workflows.engine import ( + RunState, + WorkflowDefinition, + WorkflowEngine, + ) + + monkeypatch.setattr( + RunState, "_supports_descriptor_access", staticmethod(lambda: False) + ) + definition = WorkflowDefinition.from_string( + """ +schema_version: "1.0" +workflow: + id: portable-storage + name: Portable Storage +inputs: + verdict: + type: string + default: "" +steps: + - id: review + type: gate + message: Review + verdict_input: verdict +""" + ) + engine = WorkflowEngine(project_dir) + + state = engine.execute(definition, run_id="portable") + + assert state.status == RunStatus.PAUSED + resumed = engine.resume("portable", {"verdict": "approve"}) + assert resumed.status == RunStatus.COMPLETED + assert engine.list_runs()[0]["run_id"] == "portable" + + +def test_portable_storage_fallback_rechecks_artifact_symlinks( + project_dir, tmp_path, monkeypatch +): + from specify_cli.workflows.engine import RunState + + monkeypatch.setattr( + RunState, "_supports_descriptor_access", staticmethod(lambda: False) + ) + state = _state(project_dir) + state.save() + artifact = RunState._run_artifact(project_dir, "run-1", "state.json") + external = tmp_path / "state.json" + external.write_text("external state", encoding="utf-8") + + with RunState._open_run_directory(project_dir, "run-1") as run_fd: + artifact.unlink() + artifact.symlink_to(external) + with pytest.raises(ValueError, match="symlinked workflow run artifact"): + RunState._open_artifact( + artifact, "r", dir_fd=run_fd, encoding="utf-8" + ) + + assert external.read_text(encoding="utf-8") == "external state" + + +def test_load_rejects_symlinked_run_directory(project_dir, tmp_path): + from specify_cli.workflows.engine import RunState + + outside = tmp_path / "outside" + outside.mkdir() + (outside / "state.json").write_text( + json.dumps({"run_id": "run-1", "workflow_id": "workflow", "status": "paused"}), + encoding="utf-8", + ) + runs = project_dir / ".specify" / "workflows" / "runs" + runs.mkdir(parents=True) + (runs / "run-1").symlink_to(outside, target_is_directory=True) + + with pytest.raises(ValueError, match="symlinked workflow run path"): + RunState.load("run-1", project_dir) + + +def test_load_keeps_open_run_directory_when_path_is_replaced( + project_dir, tmp_path, monkeypatch +): + from specify_cli.workflows.engine import RunState + + state = _state(project_dir) + state.save() + run_dir = project_dir / ".specify" / "workflows" / "runs" / "run-1" + detached_dir = tmp_path / "detached-run" + outside = tmp_path / "outside" + outside.mkdir() + outside_state = outside / "state.json" + outside_state.write_text( + json.dumps({ + "run_id": "run-1", + "workflow_id": "external", + "status": "paused", + }), + encoding="utf-8", + ) + original_open = RunState._open_artifact + swapped = False + + def _replace_run_directory(path, mode, *, dir_fd, **kwargs): + nonlocal swapped + if path.name == "state.json" and not swapped: + swapped = True + os.rename(run_dir, detached_dir) + run_dir.symlink_to(outside, target_is_directory=True) + return original_open(path, mode, dir_fd=dir_fd, **kwargs) + + monkeypatch.setattr( + RunState, "_open_artifact", staticmethod(_replace_run_directory) + ) + + loaded = RunState.load("run-1", project_dir) + + assert loaded.workflow_id == "workflow" + assert json.loads(outside_state.read_text(encoding="utf-8"))["workflow_id"] == "external" + + +def test_save_keeps_open_run_directory_when_path_is_replaced( + project_dir, tmp_path, monkeypatch +): + from specify_cli.workflows.engine import RunState + + state = _state(project_dir) + state.save() + run_dir = project_dir / ".specify" / "workflows" / "runs" / "run-1" + detached_dir = tmp_path / "detached-run" + outside = tmp_path / "outside" + outside.mkdir() + outside_state = outside / "state.json" + outside_state.write_text("external state", encoding="utf-8") + original_write = RunState._atomic_write_json + swapped = False + + def _replace_run_directory(path, data, *, dir_fd): + nonlocal swapped + if path.name == "state.json" and not swapped: + swapped = True + os.rename(run_dir, detached_dir) + run_dir.symlink_to(outside, target_is_directory=True) + return original_write(path, data, dir_fd=dir_fd) + + monkeypatch.setattr( + RunState, "_atomic_write_json", staticmethod(_replace_run_directory) + ) + + state.save() + + assert outside_state.read_text(encoding="utf-8") == "external state" + assert json.loads((detached_dir / "state.json").read_text(encoding="utf-8"))[ + "run_id" + ] == "run-1" + + +@pytest.mark.parametrize("artifact", ["state.json", "inputs.json"]) +def test_run_state_rejects_symlinked_persisted_artifacts( + project_dir, tmp_path, artifact +): + state = _state(project_dir) + run_dir = _run_dir(project_dir) + target = tmp_path / artifact + target.write_text("{}", encoding="utf-8") + (run_dir / artifact).symlink_to(target) + + if artifact == "state.json": + with pytest.raises(ValueError, match="symlinked workflow run artifact"): + type(state).load("run-1", project_dir) + else: + state_data = { + "run_id": "run-1", + "workflow_id": "workflow", + "status": "paused", + } + (run_dir / "state.json").write_text(json.dumps(state_data), encoding="utf-8") + with pytest.raises(ValueError, match="symlinked workflow run artifact"): + type(state).load("run-1", project_dir) + + +@pytest.mark.parametrize("artifact, operation", [ + ("state.json", "save"), + ("inputs.json", "save"), + ("log.jsonl", "log"), +]) +def test_run_state_rejects_symlinked_artifact_writes( + project_dir, tmp_path, artifact, operation +): + from specify_cli.workflows._execution import CheckpointError + + state = _state(project_dir) + run_dir = _run_dir(project_dir) + target = tmp_path / artifact + target.write_text("", encoding="utf-8") + (run_dir / artifact).symlink_to(target) + + expected = CheckpointError if operation == "save" else ValueError + with pytest.raises(expected, match="symlinked workflow run artifact"): + if operation == "save": + state.save() + else: + state.append_log({"event": "test"}) + + +def test_workflow_snapshot_artifact_is_checked(project_dir, tmp_path): + from specify_cli.workflows.engine import RunState + + run_dir = _run_dir(project_dir) + target = tmp_path / "workflow.yml" + target.write_text("workflow: {}", encoding="utf-8") + (run_dir / "workflow.yml").symlink_to(target) + + with pytest.raises(ValueError, match="symlinked workflow run artifact"): + RunState._run_artifact(project_dir, "run-1", "workflow.yml") + + +def test_execute_rejects_symlinked_workflow_snapshot(project_dir, tmp_path): + from specify_cli.workflows.engine import WorkflowDefinition, WorkflowEngine + + run_dir = _run_dir(project_dir) + target = tmp_path / "workflow.yml" + target.write_text("external workflow", encoding="utf-8") + (run_dir / "workflow.yml").symlink_to(target) + + definition = WorkflowDefinition({ + "workflow": {"id": "workflow", "name": "Workflow"}, + "steps": [], + }) + with pytest.raises(ValueError, match="symlinked workflow run artifact"): + WorkflowEngine(project_dir).execute(definition, run_id="run-1") + + assert target.read_text(encoding="utf-8") == "external workflow" + + +def test_resume_rejects_symlinked_workflow_snapshot(project_dir, tmp_path): + from specify_cli.workflows.engine import WorkflowEngine + + state = _state(project_dir) + state.save() + run_dir = _run_dir(project_dir) + target = tmp_path / "workflow.yml" + target.write_text( + "workflow: {id: workflow, name: Workflow}\n" + "steps:\n - {id: review, type: gate, message: Review}\n", + encoding="utf-8", + ) + (run_dir / "workflow.yml").symlink_to(target) + + with pytest.raises(ValueError, match="symlinked workflow run artifact"): + WorkflowEngine(project_dir).resume("run-1") + + assert target.read_text(encoding="utf-8").startswith("workflow:") diff --git a/tests/test_workflows.py b/tests/test_workflows.py index a4ea5d1786..84c9b9b420 100644 --- a/tests/test_workflows.py +++ b/tests/test_workflows.py @@ -9091,8 +9091,11 @@ def test_list_skips_invalid_utf8_with_valid_sibling(self, project_dir): assert runs[0]["workflow_id"] == "good-run-utf8" def test_list_skips_oserror_with_valid_sibling(self, project_dir, monkeypatch): - import builtins - from specify_cli.workflows.engine import WorkflowEngine, WorkflowDefinition + from specify_cli.workflows.engine import ( + RunState, + WorkflowDefinition, + WorkflowEngine, + ) runs_dir = project_dir / ".specify" / "workflows" / "runs" bad_dir = runs_dir / "bad-oserror" @@ -9100,14 +9103,14 @@ def test_list_skips_oserror_with_valid_sibling(self, project_dir, monkeypatch): state_file = bad_dir / "state.json" state_file.write_text('{"run_id": "bad"}', encoding="utf-8") - original_open = builtins.open + original_open = RunState._open_artifact - def _mock_open(path, *args, **kwargs): - if str(path).endswith("state.json") and "bad-oserror" in str(path): + def _mock_open(path, mode, *, dir_fd, **kwargs): + if path.name == "state.json" and "bad-oserror" in str(path): raise OSError("permission denied") - return original_open(path, *args, **kwargs) + return original_open(path, mode, dir_fd=dir_fd, **kwargs) - monkeypatch.setattr(builtins, "open", _mock_open) + monkeypatch.setattr(RunState, "_open_artifact", staticmethod(_mock_open)) yaml_str = """ schema_version: "1.0" diff --git a/workflows/ARCHITECTURE.md b/workflows/ARCHITECTURE.md index 6382e0c35e..3e2266e9f2 100644 --- a/workflows/ARCHITECTURE.md +++ b/workflows/ARCHITECTURE.md @@ -280,6 +280,17 @@ When `specify workflow add ` installs from catalog, it downloads the workflo | Project catalogs | `.specify/workflow-catalogs.yml` | YAML | Project-level catalog sources | | User catalogs | `~/.specify/workflow-catalogs.yml` | YAML | User-level catalog sources | +### Persisted-run storage boundary + +The `.specify/workflows/runs` tree is a project-owned persistence boundary. +Every run directory and persisted artifact below it must be a real filesystem +object rather than a symlink. `RunState` enforces this policy before reading or +writing `state.json`, `inputs.json`, `workflow.yml`, or `log.jsonl`; artifact +opens also use no-follow flags and directory descriptors where supported. +Other platforms recheck path components before access and use atomic replacement +for state writes; that fallback rejects existing symlinks but cannot prevent +concurrent path replacement. + ## Module Structure ```text diff --git a/workflows/README.md b/workflows/README.md index c952b4ad4b..ce552a60c6 100644 --- a/workflows/README.md +++ b/workflows/README.md @@ -520,6 +520,12 @@ specify workflow resume Run states: `created` → `running` → `completed` | `paused` | `failed` | `aborted` +Persisted-run storage rejects every symlink below `.specify/workflows/runs`, +including per-run directories and artifact files. This applies consistently to +state, inputs, workflow snapshots, and logs for both reads and writes. Workflow +storage cannot therefore be relocated through symlinks; move the files and +directories into the project-owned runs directory instead. + ## Catalog Management Workflows are discovered through catalogs. By default, Spec Kit uses the official and community catalogs: