Load registry from disk or create default.
(self)
| 649 | return False |
| 650 | |
| 651 | def _load(self) -> dict[str, Any]: |
| 652 | """Load registry from disk or create default.""" |
| 653 | default_registry: dict[str, Any] = {"schema_version": self.SCHEMA_VERSION, "steps": {}} |
| 654 | # Defense-in-depth: refuse to read the registry if any parent directory |
| 655 | # under .specify/workflows/steps is a symlink, which could redirect the |
| 656 | # read outside the project root. |
| 657 | if self._has_symlinked_parent(): |
| 658 | return default_registry |
| 659 | # Defense-in-depth: also refuse to read a symlinked registry file, |
| 660 | # which could redirect the read outside the project root. |
| 661 | if self.registry_path.is_symlink(): |
| 662 | return default_registry |
| 663 | if self.registry_path.exists(): |
| 664 | try: |
| 665 | with open(self.registry_path, encoding="utf-8") as f: |
| 666 | data = json.load(f) |
| 667 | # Validate shape: must be a dict with a dict "steps" field |
| 668 | if not isinstance(data, dict): |
| 669 | return default_registry |
| 670 | if not isinstance(data.get("steps"), dict): |
| 671 | data["steps"] = {} |
| 672 | return data |
| 673 | except (json.JSONDecodeError, ValueError, OSError, UnicodeError): |
| 674 | return default_registry |
| 675 | return default_registry |
| 676 | |
| 677 | def save(self) -> None: |
| 678 | """Persist registry to disk. |
no test coverage detected