Prevent workflow commands from targeting reserved storage

Workflow install and removal paths are derived from workflow IDs before any catalog download, local copy, or directory deletion. Validate that IDs are single workflow-id path segments and reject names reserved for workflow runtime storage so commands cannot target .specify/workflows/runs or .specify/workflows/steps.

Author: darion-yaphet

src/specify_cli/workflows/_commands.py

@@ -10,6 +10,7 @@
 import contextlib
 import json
 import os
+import re
 import sys
 from pathlib import Path
 from typing import Any
@@ -90,6 +91,22 @@ def _reject_unsafe_workflow_storage(project_root: Path) -> None:
     )
 
 
+_WORKFLOW_ID_PATTERN = re.compile(r"^[a-z0-9](?:[a-z0-9-]*[a-z0-9])?$")
+_RESERVED_WORKFLOW_IDS: frozenset[str] = frozenset({"runs", "steps"})
+
+
+def _validate_workflow_id_or_exit(workflow_id: str) -> None:
+    """Validate that ``workflow_id`` is a safe installed-workflow directory name."""
+    if (
+        workflow_id in _RESERVED_WORKFLOW_IDS
+        or not _WORKFLOW_ID_PATTERN.match(workflow_id)
+    ):
+        console.print(
+            f"[red]Error:[/red] Invalid workflow ID: {_escape_markup(repr(workflow_id))}"
+        )
+        raise typer.Exit(1)
+
+
 def _safe_workflow_id_dir(workflows_dir: Path, workflow_id: str) -> Path:
     """Validate the per-id install directory before any write and return it.
 
@@ -100,6 +117,8 @@ def _safe_workflow_id_dir(workflows_dir: Path, workflow_id: str) -> Path:
 
     - an ``<id>`` that is a symlink or an existing non-directory
       (the latter would otherwise make ``mkdir`` raise);
+    - an ``<id>`` that is not a single workflow-id path segment or collides
+      with internal workflow storage directories;
     - an ``<id>`` that escapes ``workflows_dir`` (path traversal);
     - an ``<id>/workflow.yml`` leaf that is a symlink or an existing
       non-file (either would otherwise make the later write/copy raise).
@@ -109,6 +128,8 @@ def _safe_workflow_id_dir(workflows_dir: Path, workflow_id: str) -> Path:
     ``workflow_id`` is markup-escaped in output to avoid Rich markup injection.
     """
     safe_id = _escape_markup(workflow_id)
+    _validate_workflow_id_or_exit(workflow_id)
+
     dest_dir = workflows_dir / workflow_id
     _reject_unsafe_dir(dest_dir, f".specify/workflows/{safe_id}")
     try:
@@ -801,17 +822,54 @@ def workflow_remove(
     from .catalog import WorkflowRegistry
 
     project_root = _require_specify_project()
+    workflows_dir = project_root / ".specify" / "workflows"
+    _validate_workflow_id_or_exit(workflow_id)
+
     registry = WorkflowRegistry(project_root)
 
     if not registry.is_installed(workflow_id):
         console.print(f"[red]Error:[/red] Workflow '{workflow_id}' is not installed")
         raise typer.Exit(1)
 
     # Remove workflow files
-    workflow_dir = project_root / ".specify" / "workflows" / workflow_id
+    workflow_dir_unresolved = workflows_dir / workflow_id
+    safe_id = _escape_markup(workflow_id)
+    if workflow_dir_unresolved.is_symlink():
+        console.print(
+            f"[red]Error:[/red] Refusing to remove symlinked "
+            f".specify/workflows/{safe_id}"
+        )
+        raise typer.Exit(1)
+
+    workflow_dir = workflow_dir_unresolved.resolve()
+    try:
+        rel_parts = workflow_dir.relative_to(workflows_dir.resolve()).parts
+    except ValueError:
+        console.print(
+            f"[red]Error:[/red] Invalid workflow ID: {_escape_markup(repr(workflow_id))}"
+        )
+        raise typer.Exit(1)
+    if rel_parts != (workflow_id,):
+        console.print(
+            f"[red]Error:[/red] Invalid workflow ID: {_escape_markup(repr(workflow_id))}"
+        )
+        raise typer.Exit(1)
+
+    if workflow_dir.exists() and not workflow_dir.is_dir():
+        console.print(
+            f"[red]Error:[/red] .specify/workflows/{safe_id} exists but is not a directory"
+        )
+        raise typer.Exit(1)
+
     if workflow_dir.exists():
         import shutil
-        shutil.rmtree(workflow_dir)
+        try:
+            shutil.rmtree(workflow_dir)
+        except OSError as exc:
+            console.print(
+                f"[red]Error:[/red] Failed to remove workflow directory {workflow_dir}: {exc}"
+            )
+            raise typer.Exit(1)
 
     registry.remove(workflow_id)
     console.print(f"[green]✓[/green] Workflow '{workflow_id}' removed")

tests/test_workflows.py

@@ -5303,6 +5303,95 @@ def test_remove_rejects_symlinked_steps_base_dir(self, project_dir, monkeypatch)
         assert "Refusing to use symlinked step directory" in result.output
 
 
+class TestWorkflowRemoveGuard:
+    def test_remove_rejects_traversal_registry_key(self, project_dir, monkeypatch):
+        """A corrupted registry key must not let remove delete outside workflows/."""
+        from typer.testing import CliRunner
+        from specify_cli import app
+        from specify_cli.workflows.catalog import WorkflowRegistry
+
+        registry = WorkflowRegistry(project_dir)
+        registry.add("../outside", {"name": "Bad"})
+        outside = project_dir / ".specify" / "outside"
+        outside.mkdir()
+        sentinel = outside / "keep.txt"
+        sentinel.write_text("keep", encoding="utf-8")
+
+        monkeypatch.chdir(project_dir)
+        result = CliRunner().invoke(app, ["workflow", "remove", "../outside"])
+
+        assert result.exit_code != 0
+        assert "Invalid workflow ID" in result.output
+        assert sentinel.read_text(encoding="utf-8") == "keep"
+
+    @pytest.mark.parametrize("workflow_id", ["runs", "steps"])
+    def test_remove_rejects_reserved_storage_ids(
+        self, project_dir, monkeypatch, workflow_id
+    ):
+        """Reserved workflow storage directories must never be removable workflows."""
+        from typer.testing import CliRunner
+        from specify_cli import app
+        from specify_cli.workflows.catalog import WorkflowRegistry
+
+        registry = WorkflowRegistry(project_dir)
+        registry.add(workflow_id, {"name": "Bad"})
+        reserved_dir = project_dir / ".specify" / "workflows" / workflow_id
+        reserved_dir.mkdir(exist_ok=True)
+        sentinel = reserved_dir / "keep.txt"
+        sentinel.write_text("keep", encoding="utf-8")
+
+        monkeypatch.chdir(project_dir)
+        result = CliRunner().invoke(app, ["workflow", "remove", workflow_id])
+
+        assert result.exit_code != 0
+        assert "Invalid workflow ID" in result.output
+        assert sentinel.read_text(encoding="utf-8") == "keep"
+
+    @pytest.mark.skipif(not hasattr(os, "symlink"), reason="symlinks are unavailable")
+    def test_remove_refuses_symlinked_workflow_dir(self, project_dir, monkeypatch):
+        """A symlinked workflow directory must not let remove delete its target."""
+        from typer.testing import CliRunner
+        from specify_cli import app
+        from specify_cli.workflows.catalog import WorkflowRegistry
+
+        registry = WorkflowRegistry(project_dir)
+        registry.add("test-wf", {"name": "Test"})
+        outside = project_dir / "outside-workflow-remove-target"
+        outside.mkdir(exist_ok=True)
+        sentinel = outside / "keep.txt"
+        sentinel.write_text("keep", encoding="utf-8")
+        (project_dir / ".specify" / "workflows" / "test-wf").symlink_to(
+            outside, target_is_directory=True
+        )
+
+        monkeypatch.chdir(project_dir)
+        result = CliRunner().invoke(app, ["workflow", "remove", "test-wf"])
+
+        assert result.exit_code != 0
+        assert "symlinked .specify/workflows/test-wf" in result.output
+        assert sentinel.read_text(encoding="utf-8") == "keep"
+        assert WorkflowRegistry(project_dir).is_installed("test-wf")
+
+    def test_remove_refuses_non_directory_workflow_path(self, project_dir, monkeypatch):
+        """A file at the workflow path must fail cleanly instead of crashing."""
+        from typer.testing import CliRunner
+        from specify_cli import app
+        from specify_cli.workflows.catalog import WorkflowRegistry
+
+        registry = WorkflowRegistry(project_dir)
+        registry.add("test-wf", {"name": "Test"})
+        workflow_path = project_dir / ".specify" / "workflows" / "test-wf"
+        workflow_path.write_text("not a directory", encoding="utf-8")
+
+        monkeypatch.chdir(project_dir)
+        result = CliRunner().invoke(app, ["workflow", "remove", "test-wf"])
+
+        assert result.exit_code != 0
+        assert "exists but is not a directory" in result.output
+        assert workflow_path.read_text(encoding="utf-8") == "not a directory"
+        assert WorkflowRegistry(project_dir).is_installed("test-wf")
+
+
 class TestWorkflowAddSymlinkGuard:
     @pytest.mark.skipif(not hasattr(os, "symlink"), reason="symlinks are unavailable")
     def test_add_refuses_symlinked_specify(self, temp_dir, monkeypatch):
@@ -5439,6 +5528,34 @@ def test_safe_workflow_id_dir_escapes_markup_in_invalid_id(self, temp_dir, capsy
         # Literal bracketed text survives; Rich did not consume it as a tag.
         assert "[red]evil[/red]" in out
 
+    @pytest.mark.parametrize(
+        "workflow_id",
+        [
+            "runs",
+            "steps",
+            "nested/workflow",
+            "nested\\workflow",
+            "bad id",
+            " bad-id",
+            "bad-id ",
+        ],
+    )
+    def test_safe_workflow_id_dir_rejects_reserved_or_non_segment_ids(
+        self, temp_dir, workflow_id, capsys
+    ):
+        """Install IDs must not collide with workflow internals or create nested paths."""
+        import typer
+        from specify_cli.workflows._commands import _safe_workflow_id_dir
+
+        workflows_dir = temp_dir / ".specify" / "workflows"
+        workflows_dir.mkdir(parents=True)
+
+        with pytest.raises(typer.Exit):
+            _safe_workflow_id_dir(workflows_dir, workflow_id)
+
+        assert "Invalid workflow ID" in capsys.readouterr().out
+        assert not (workflows_dir / workflow_id).exists()
+
     @pytest.mark.skipif(not hasattr(os, "symlink"), reason="symlinks are unavailable")
     def test_list_refuses_symlinked_runs_dir(self, temp_dir, monkeypatch):
         """workflow commands using the project shim must refuse symlinked run storage."""