Class StepRegistry

src/specify_cli/workflows/catalog.py:626–729

Manages the registry of installed custom step types. Tracks installed step types and their metadata in ``.specify/workflows/steps/step-registry.json``.

class StepRegistry:
    """Manages the registry of installed custom step types.

    Tracks installed step types and their metadata in
    ``.specify/workflows/steps/step-registry.json``.
    """

    REGISTRY_FILE = "step-registry.json"
    SCHEMA_VERSION = "1.0"

    def __init__(self, project_root: Path) -> None:
        self.project_root = project_root
        self.steps_dir = project_root / ".specify" / "workflows" / "steps"
        self.registry_path = self.steps_dir / self.REGISTRY_FILE
        self.data = self._load()

    def _has_symlinked_parent(self) -> bool:
        """Return True if any directory under .specify/workflows/steps is a symlink."""
        current = self.project_root
        for part in (".specify", "workflows", "steps"):
            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, "steps": {}}
        # Defense-in-depth: refuse to read the registry if any parent directory
        # under .specify/workflows/steps is a symlink, which could redirect the
        # read outside the project root.
        if self._has_symlinked_parent():
            return default_registry
        # Defense-in-depth: also refuse to read a symlinked registry file,
        # which could redirect the read outside the project root.
        if self.registry_path.is_symlink():
            return default_registry
        if self.registry_path.exists():
            try:
                with open(self.registry_path, encoding="utf-8") as f:
                    data = json.load(f)
                # Validate shape: must be a dict with a dict "steps" field
                if not isinstance(data, dict):
                    return default_registry
                if not isinstance(data.get("steps"), dict):
                    data["steps"] = {}
                return data
            except (json.JSONDecodeError, ValueError, OSError, UnicodeError):
                return default_registry
        return default_registry

    def save(self) -> None:
        """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``.
        """