Harden step_id validation against path-segment tricks; raise on StepRegistry.save() OSError

Author: Copilot

src/specify_cli/__init__.py

@@ -5938,6 +5938,27 @@ def workflow_step_list():
 _RESERVED_STEP_IDS: frozenset[str] = frozenset({".cache", "step-registry.json"})
 
 
+def _validate_step_id_or_exit(step_id: str) -> None:
+    """Validate that ``step_id`` is a single safe path component.
+
+    Rejects empty strings, path separators, ``.``/``..`` components, dotfile
+    prefixes, and reserved names. Exits with code 1 on failure.
+    """
+    if (
+        not step_id
+        or "/" in step_id
+        or "\\" in step_id
+        or step_id in (".", "..")
+        or step_id.startswith(".")
+        or step_id in _RESERVED_STEP_IDS
+    ):
+        console.print(
+            f"[red]Error:[/red] Invalid step id '{step_id}': must be a single safe "
+            "path component (no separators, no leading dot, not a reserved name)"
+        )
+        raise typer.Exit(1)
+
+
 @workflow_step_app.command("add")
 def workflow_step_add(
     step_id: str = typer.Argument(..., help="Step type ID from catalog"),
@@ -6027,21 +6048,19 @@ def _safe_fetch(url: str) -> bytes:
                 raise ValueError(f"Redirect to URL with no hostname: {final_url}")
             return resp.read()
 
+    _validate_step_id_or_exit(step_id)
+
     steps_base_dir = (project_root / ".specify" / "workflows" / "steps").resolve()
     step_dir = (steps_base_dir / step_id).resolve()
+    # Defense-in-depth: ensure the resolved directory is a direct child of
+    # steps_base_dir even after symlink resolution.
     try:
-        step_dir.relative_to(steps_base_dir)
+        rel_parts = step_dir.relative_to(steps_base_dir).parts
     except ValueError:
         console.print(f"[red]Error:[/red] Invalid step id '{step_id}'")
         raise typer.Exit(1)
-
-    # Reject IDs that collide with internal names used under steps_base_dir
-    # (dotfiles, the cache dir, and the registry filename) to prevent
-    # corrupting caching or registry persistence.
-    if step_id.startswith(".") or step_id in _RESERVED_STEP_IDS:
-        console.print(
-            f"[red]Error:[/red] '{step_id}' is a reserved name and cannot be used as a step ID"
-        )
+    if rel_parts != (step_id,):
+        console.print(f"[red]Error:[/red] Invalid step id '{step_id}'")
         raise typer.Exit(1)
 
     import shutil
@@ -6183,16 +6202,24 @@ def workflow_step_remove(
         console.print("[red]Error:[/red] Not a spec-kit project (no .specify/ directory)")
         raise typer.Exit(1)
 
+    _validate_step_id_or_exit(step_id)
+
     registry = StepRegistry(project_root)
     in_registry = registry.is_installed(step_id)
 
     steps_base_dir = (project_root / ".specify" / "workflows" / "steps").resolve()
     step_dir = (steps_base_dir / step_id).resolve()
+    # Defense-in-depth: even though _validate_step_id_or_exit rejects path
+    # separators, ensure that the resolved directory is a single child of
+    # steps_base_dir and is not steps_base_dir itself.
     try:
-        step_dir.relative_to(steps_base_dir)
+        rel_parts = step_dir.relative_to(steps_base_dir).parts
     except ValueError:
         console.print(f"[red]Error:[/red] Invalid step id '{step_id}'")
         raise typer.Exit(1)
+    if rel_parts != (step_id,):
+        console.print(f"[red]Error:[/red] Invalid step id '{step_id}'")
+        raise typer.Exit(1)
 
     dir_exists = step_dir.exists()
 

src/specify_cli/workflows/catalog.py

@@ -616,10 +616,20 @@ def _load(self) -> dict[str, Any]:
         return default_registry
 
     def save(self) -> None:
-        """Persist registry to disk."""
-        self.steps_dir.mkdir(parents=True, exist_ok=True)
-        with open(self.registry_path, "w", encoding="utf-8") as f:
-            json.dump(self.data, f, indent=2)
+        """Persist registry to disk.
+
+        Raises ``StepValidationError`` with a clear message on filesystem
+        errors (read-only fs, permission denied, ...) so callers can surface
+        a clean error to the user rather than an unhandled ``OSError``.
+        """
+        try:
+            self.steps_dir.mkdir(parents=True, exist_ok=True)
+            with open(self.registry_path, "w", encoding="utf-8") as f:
+                json.dump(self.data, f, indent=2)
+        except OSError as exc:
+            raise StepValidationError(
+                f"Failed to write step registry at {self.registry_path}: {exc}"
+            ) from exc
 
     def add(self, step_id: str, metadata: dict[str, Any]) -> None:
         """Add or update an installed step entry."""