From cac5300f89c8bed41198591edd88a0a50cc75c2f Mon Sep 17 00:00:00 2001 From: marcelsafin <179933638+marcelsafin@users.noreply.github.com> Date: Sat, 11 Jul 2026 00:43:11 +0200 Subject: [PATCH] fix(workflows): symlink guards and shape validation in workflow registry, dev-dir file check - WorkflowRegistry now mirrors StepRegistry: _load refuses symlinked parents/registry file and normalizes a non-dict workflows field; save() rejects symlinked paths before writing. - workflow add --dev requires workflow.yml to be a regular file so a directory named workflow.yml gets the documented CLI error instead of an uncaught IsADirectoryError. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- src/specify_cli/workflows/_commands.py | 2 +- src/specify_cli/workflows/catalog.py | 37 ++++++++++++++++++++++--- tests/test_workflows.py | 38 ++++++++++++++++++++++++++ 3 files changed, 72 insertions(+), 5 deletions(-) diff --git a/src/specify_cli/workflows/_commands.py b/src/specify_cli/workflows/_commands.py index 334c7e254..03d5a4b8a 100644 --- a/src/specify_cli/workflows/_commands.py +++ b/src/specify_cli/workflows/_commands.py @@ -703,7 +703,7 @@ def workflow_add( return if dev_path.is_dir(): dev_wf_file = dev_path / "workflow.yml" - if not dev_wf_file.exists(): + if not dev_wf_file.is_file(): console.print(f"[red]Error:[/red] No workflow.yml found in {_escape_markup(source)}") raise typer.Exit(1) _validate_and_install_local(dev_wf_file, str(dev_path)) diff --git a/src/specify_cli/workflows/catalog.py b/src/specify_cli/workflows/catalog.py index 87fe1ca0b..092b4b6bf 100644 --- a/src/specify_cli/workflows/catalog.py +++ b/src/specify_cli/workflows/catalog.py @@ -72,19 +72,48 @@ class WorkflowRegistry: self.registry_path = self.workflows_dir / self.REGISTRY_FILE self.data = self._load() + def _has_symlinked_parent(self) -> bool: + """Return True if any directory under .specify/workflows is a symlink.""" + current = self.project_root + for part in (".specify", "workflows"): + current = current / part + if current.is_symlink(): + return True + return False + def _load(self) -> dict[str, Any]: """Load registry from disk or create default.""" + default_registry: dict[str, Any] = { + "schema_version": self.SCHEMA_VERSION, + "workflows": {}, + } + # Defense-in-depth: refuse to read through symlinked parents or a + # symlinked registry file (mirrors StepRegistry._load). + if self._has_symlinked_parent() or self.registry_path.is_symlink(): + return default_registry if self.registry_path.exists(): try: with open(self.registry_path, encoding="utf-8") as f: - return json.load(f) - except (json.JSONDecodeError, ValueError): + data = json.load(f) + # Validate shape: must be a dict with a dict "workflows" field. + if not isinstance(data, dict): + return default_registry + if not isinstance(data.get("workflows"), dict): + data["workflows"] = {} + return data + except (json.JSONDecodeError, ValueError, OSError, UnicodeError): # Corrupted registry file — reset to default - return {"schema_version": self.SCHEMA_VERSION, "workflows": {}} - return {"schema_version": self.SCHEMA_VERSION, "workflows": {}} + return default_registry + return default_registry def save(self) -> None: """Persist registry to disk atomically.""" + # Refuse to write through symlinked parents (mirrors StepRegistry.save + # and the CLI-level _reject_unsafe_dir guard). + if self._has_symlinked_parent() or self.registry_path.is_symlink(): + raise OSError( + "Refusing to write workflow registry through a symlinked path." + ) self.workflows_dir.mkdir(parents=True, exist_ok=True) # Unique, exclusive temp then replace: a failed dump cannot truncate # the registry, a pre-created symlink cannot redirect the write, and diff --git a/tests/test_workflows.py b/tests/test_workflows.py index 810a94dc3..d8c702b18 100644 --- a/tests/test_workflows.py +++ b/tests/test_workflows.py @@ -7563,6 +7563,44 @@ steps: registry.add("align-wf", {"version": "1.0.0", "source": "catalog"}) assert registry.get("align-wf")["version"] == "1.0.0" + def test_registry_load_normalizes_malformed_workflows_field(self, project_dir): + from specify_cli.workflows.catalog import WorkflowRegistry + + registry = WorkflowRegistry(project_dir) + registry.workflows_dir.mkdir(parents=True, exist_ok=True) + registry.registry_path.write_text('{"workflows": "broken"}', encoding="utf-8") + fresh = WorkflowRegistry(project_dir) + assert fresh.get("anything") is None + fresh.add("align-wf", {"version": "1.0.0", "source": "catalog"}) + assert fresh.get("align-wf")["version"] == "1.0.0" + + def test_registry_save_refuses_symlinked_parent(self, project_dir, tmp_path): + from specify_cli.workflows.catalog import WorkflowRegistry + + outside = tmp_path / "outside-specify" + outside.mkdir() + specify_dir = project_dir / ".specify" + if specify_dir.exists(): + shutil.rmtree(specify_dir) + specify_dir.symlink_to(outside) + registry = WorkflowRegistry(project_dir) + with pytest.raises(OSError, match="symlink"): + registry.add("align-wf", {"version": "1.0.0", "source": "catalog"}) + assert not (outside / "workflows").exists() + + def test_add_dev_dir_with_workflow_yml_directory_errors_cleanly(self, project_dir, monkeypatch): + from typer.testing import CliRunner + from specify_cli import app + + monkeypatch.chdir(project_dir) + dev_dir = project_dir / "dev-wf" + (dev_dir / "workflow.yml").mkdir(parents=True) + runner = CliRunner() + result = runner.invoke(app, ["workflow", "add", "--dev", str(dev_dir)]) + assert result.exit_code != 0 + assert result.exception is None or isinstance(result.exception, SystemExit) + assert "No workflow.yml found" in result.output + def test_registry_save_failure_preserves_file_on_disk(self, project_dir, monkeypatch): """A failed dump must not truncate the persisted registry.""" from specify_cli.workflows.catalog import WorkflowRegistry