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>
This commit is contained in:
marcelsafin
2026-07-11 00:43:11 +02:00
parent eabce4efe1
commit cac5300f89
3 changed files with 72 additions and 5 deletions

View File

@@ -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))

View File

@@ -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

View File

@@ -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