diff --git a/src/specify_cli/workflows/step/_helpers.py b/src/specify_cli/workflows/step/_helpers.py index 6b8fdd8576..e092f3a223 100644 --- a/src/specify_cli/workflows/step/_helpers.py +++ b/src/specify_cli/workflows/step/_helpers.py @@ -47,3 +47,18 @@ def _resolve_steps_base_dir_or_exit(project_root: cli.Path) -> cli.Path: except StepInstallError as exc: cli.console.print(f"[red]Error:[/red] {cli._escape_markup(str(exc))}") raise cli.typer.Exit(1) from exc + + +def _load_step_registry_or_exit(project_root: cli.Path): + """Build the step registry, turning an OSError into a clean CLI error. + + The registry fails loudly when the file is symlinked or corrupted, so + surface that as a printed error rather than an unhandled traceback. + """ + from .catalog import StepRegistry + + try: + return StepRegistry(project_root) + except OSError as exc: + cli.console.print(f"[red]Error:[/red] {cli._escape_markup(str(exc))}") + raise cli.typer.Exit(1) from exc diff --git a/src/specify_cli/workflows/step/catalog/_domain.py b/src/specify_cli/workflows/step/catalog/_domain.py index d1f9ecf75c..d058afffba 100644 --- a/src/specify_cli/workflows/step/catalog/_domain.py +++ b/src/specify_cli/workflows/step/catalog/_domain.py @@ -100,29 +100,45 @@ def _has_symlinked_parent(self) -> bool: return False def _load(self) -> dict[str, Any]: - """Load registry from disk or create default.""" + """Load registry from disk or create default. + + Raises ``OSError`` when the registry exists but cannot be read or is + not a valid registry document, instead of silently handing back an + empty registry and hiding the corruption from the caller. + """ 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 + # read outside the project root, or if the registry file itself is one. + if self._has_symlinked_parent() or self.registry_path.is_symlink(): + raise OSError( + f"Refusing to read step registry at {self.registry_path}: " + "a parent directory or the registry file itself is a symlink" + ) 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 + except OSError as exc: + raise OSError( + f"Failed to read step registry at {self.registry_path}: {exc}" + ) from exc + except (json.JSONDecodeError, ValueError, UnicodeError) as exc: + raise OSError( + f"Step registry at {self.registry_path} is corrupted: {exc}" + ) from exc + # Validate shape: must be a dict with a dict "steps" field + if not isinstance(data, dict): + raise OSError( + f"Step registry at {self.registry_path} is corrupted: " + "top-level value must be an object" + ) + if not isinstance(data.get("steps"), dict): + raise OSError( + f"Step registry at {self.registry_path} is corrupted: " + "'steps' must be an object" + ) + return data return default_registry def save(self) -> None: diff --git a/src/specify_cli/workflows/step/command_info.py b/src/specify_cli/workflows/step/command_info.py index cc2b023995..f80c0c50a1 100644 --- a/src/specify_cli/workflows/step/command_info.py +++ b/src/specify_cli/workflows/step/command_info.py @@ -35,12 +35,13 @@ def workflow_step_info( ): """Show details for a step type.""" from .. import STEP_REGISTRY - from .catalog import StepCatalog, StepCatalogError, StepRegistry + from . import _helpers as step_helpers + from .catalog import StepCatalog, StepCatalogError project_root = cli._require_specify_project() safe_step_id = cli._escape_markup(str(step_id)) - registry = StepRegistry(project_root) + registry = step_helpers._load_step_registry_or_exit(project_root) installed_meta = registry.get(step_id) # Check if it's a built-in diff --git a/src/specify_cli/workflows/step/command_list.py b/src/specify_cli/workflows/step/command_list.py index 35afdf5349..a32f223534 100644 --- a/src/specify_cli/workflows/step/command_list.py +++ b/src/specify_cli/workflows/step/command_list.py @@ -10,15 +10,15 @@ def workflow_step_list(): """List installed step types (built-in and custom).""" from .. import STEP_REGISTRY - from .catalog import StepRegistry + from . import _helpers as step_helpers project_root = cli._require_specify_project() specify_dir = project_root / ".specify" - # Read installed custom steps from registry only — no dynamic imports + # Read installed custom steps from registry only �?" no dynamic imports installed: dict = {} if specify_dir.exists(): - registry = StepRegistry(project_root) + registry = step_helpers._load_step_registry_or_exit(project_root) installed = registry.list() cli.console.print("\n[bold cyan]Installed Step Types:[/bold cyan]\n") diff --git a/src/specify_cli/workflows/step/command_remove.py b/src/specify_cli/workflows/step/command_remove.py index 52a9a08a6b..3e4931c03c 100644 --- a/src/specify_cli/workflows/step/command_remove.py +++ b/src/specify_cli/workflows/step/command_remove.py @@ -17,13 +17,15 @@ def _remove_step_locked(project_root: cli.Path, step_id: str) -> None: """ import shutil - from .catalog import StepRegistry, StepValidationError + from .catalog import StepValidationError safe_step_id = cli._escape_markup(step_id) - registry = StepRegistry(project_root) + # Directory-level symlink refusal runs first so it keeps its specific + # message; registry construction then surfaces read errors cleanly. + steps_base_dir = step_helpers._resolve_steps_base_dir_or_exit(project_root) + registry = step_helpers._load_step_registry_or_exit(project_root) in_registry = registry.is_installed(step_id) - steps_base_dir = step_helpers._resolve_steps_base_dir_or_exit(project_root) step_dir = (steps_base_dir / step_id).resolve() # Defense-in-depth: even though step_helpers._validate_step_id_or_exit rejects path # separators, ensure that the resolved directory is a single child of diff --git a/src/specify_cli/workflows/step/installer.py b/src/specify_cli/workflows/step/installer.py index de55d1f39d..84539500f8 100644 --- a/src/specify_cli/workflows/step/installer.py +++ b/src/specify_cli/workflows/step/installer.py @@ -164,6 +164,21 @@ def resolve_steps_base_dir(project_root: Path) -> Path: return steps_base_dir +def _load_step_registry(project_root: Path): + """Load the step registry, mapping read failures onto ``StepInstallError``. + + ``StepRegistry`` raises ``OSError`` for a symlinked or corrupted registry + file, while callers of this module only handle ``StepInstallError``, so + translate the error instead of letting it escape as a traceback. + """ + from .catalog import StepRegistry + + try: + return StepRegistry(project_root) + except OSError as exc: + raise StepInstallError(str(exc)) from exc + + def _resolve_step_dir(steps_base_dir: Path, step_id: str) -> Path: """Return the canonical destination directory for ``step_id``.""" step_dir = steps_base_dir / step_id @@ -474,14 +489,12 @@ def check_installable(project_root: Path, step_id: str, *, force: bool = False) uses this to reject before a download; :func:`install_step_package` re-runs the same checks as defense-in-depth. """ - from .catalog import StepRegistry - validate_step_id(step_id) steps_base_dir = resolve_steps_base_dir(project_root) step_dir = _resolve_step_dir(steps_base_dir, step_id) _reject_unsafe_destination(step_dir) _reject_builtin_collision(step_id) - registry = StepRegistry(project_root) + registry = _load_step_registry(project_root) _check_duplicate(registry, step_id, step_dir, force=force) return step_dir @@ -742,8 +755,6 @@ def install_step_package( ``source`` is exactly ``"catalog"``, ``"local"``, or ``"url"``. Returns the registry entry that was persisted. """ - from .catalog import StepRegistry - package_dir = Path(package_dir) if package_dir.is_symlink(): raise StepInstallError( @@ -774,7 +785,7 @@ def install_step_package( ) _reject_builtin_collision(step_id) - registry = StepRegistry(project_root) + registry = _load_step_registry(project_root) _check_duplicate(registry, step_id, step_dir, force=force) # Validate source and all caller-controlled metadata before creating any @@ -821,7 +832,7 @@ def install_step_package( step_dir = _resolve_step_dir(locked_base_dir, step_id) _reject_unsafe_destination(step_dir) _reject_builtin_collision(step_id) - registry = StepRegistry(project_root) + registry = _load_step_registry(project_root) _check_duplicate(registry, step_id, step_dir, force=force) _replace_install( step_dir, diff --git a/tests/test_workflows.py b/tests/test_workflows.py index bb83f3eda3..4ad4b1cb86 100644 --- a/tests/test_workflows.py +++ b/tests/test_workflows.py @@ -9876,19 +9876,20 @@ def test_persistence(self, project_dir): registry2 = StepRegistry(project_dir) assert registry2.is_installed("deploy") - def test_corrupted_registry_resets(self, project_dir): + def test_corrupted_registry_raises(self, project_dir): from specify_cli.workflows.step.catalog import StepRegistry registry = StepRegistry(project_dir) registry.steps_dir.mkdir(parents=True, exist_ok=True) registry.registry_path.write_text("not json", encoding="utf-8") - # Loading again should reset - registry2 = StepRegistry(project_dir) - assert registry2.list() == {} + # Loading again must fail loudly instead of handing back an empty + # registry that hides the corruption. + with pytest.raises(OSError, match="corrupted"): + StepRegistry(project_dir) - def test_registry_missing_steps_key_resets(self, project_dir): - """Valid JSON but missing 'steps' key should not crash add/get.""" + def test_registry_invalid_steps_key_raises(self, project_dir): + """Valid JSON but a non-dict 'steps' must fail loudly, not repair.""" from specify_cli.workflows.step.catalog import StepRegistry import json as _json @@ -9900,15 +9901,12 @@ def test_registry_missing_steps_key_resets(self, project_dir): encoding="utf-8", ) - registry2 = StepRegistry(project_dir) - # Should be safe to call add/get without KeyError - assert registry2.list() == {} - registry2.add("deploy", {"name": "Deploy", "type_key": "deploy"}) - assert registry2.is_installed("deploy") + with pytest.raises(OSError, match="'steps' must be an object"): + StepRegistry(project_dir) @pytest.mark.skipif(sys.platform == "win32", reason="chmod not reliable on Windows") - def test_registry_unreadable_file_resets(self, project_dir): - """OSError reading the registry file should fall back to default.""" + def test_registry_unreadable_file_raises(self, project_dir): + """An OSError reading the registry file must surface to the caller.""" from specify_cli.workflows.step.catalog import StepRegistry import json as _json @@ -9922,12 +9920,13 @@ def test_registry_unreadable_file_resets(self, project_dir): # Make it unreadable registry.registry_path.chmod(0o000) try: - registry2 = StepRegistry(project_dir) - assert registry2.list() == {} + with pytest.raises(OSError, match="Failed to read"): + StepRegistry(project_dir) finally: registry.registry_path.chmod(0o644) # After restoring permissions the registry is fully functional + registry2 = StepRegistry(project_dir) registry2.add("deploy", {"name": "Deploy", "type_key": "deploy"}) assert registry2.is_installed("deploy") @@ -9946,20 +9945,23 @@ def test_registry_load_refuses_symlinked_steps_dir(self, project_dir): steps_link = project_dir / ".specify" / "workflows" / "steps" steps_link.symlink_to(outside, target_is_directory=True) - registry = StepRegistry(project_dir) - assert registry.list() == {} + with pytest.raises(OSError, match="symlink"): + StepRegistry(project_dir) @pytest.mark.skipif(not hasattr(os, "symlink"), reason="symlinks are unavailable") def test_registry_save_refuses_symlinked_steps_dir(self, project_dir): """save() must refuse symlinked registry paths (defense-in-depth).""" from specify_cli.workflows.step.catalog import StepRegistry, StepValidationError + # Build the registry before swapping in the symlink so the refusal + # comes from save() itself rather than the constructor's read check. + registry = StepRegistry(project_dir) + outside = project_dir.parent / "outside-steps-save" outside.mkdir(parents=True, exist_ok=True) steps_link = project_dir / ".specify" / "workflows" / "steps" steps_link.symlink_to(outside, target_is_directory=True) - registry = StepRegistry(project_dir) with pytest.raises(StepValidationError, match="symlinked path"): registry.save() @@ -11412,3 +11414,49 @@ def boom(*args, **kwargs): assert fresh.get("align-wf")["version"] == "1.0.0" assert stat.S_IMODE(registry.registry_path.stat().st_mode) == 0o644 assert not list(registry.workflows_dir.glob("*.tmp")) + + +class TestStepRegistryAtomicSaveRegression: + """StepRegistry must fail loudly on corruption and save atomically.""" + + @pytest.mark.skipif(not hasattr(os, "symlink"), reason="symlinks are unavailable") + def test_load_raises_on_symlinked_file(self, tmp_path): + from specify_cli.workflows.step.catalog import StepRegistry + + steps_dir = tmp_path / ".specify" / "workflows" / "steps" + steps_dir.mkdir(parents=True) + registry_path = steps_dir / StepRegistry.REGISTRY_FILE + real_file = tmp_path / "real_registry.json" + real_file.write_text( + json.dumps({"schema_version": "1.0", "steps": {}}), encoding="utf-8" + ) + registry_path.symlink_to(real_file) + + with pytest.raises(OSError, match="symlink"): + StepRegistry(tmp_path) + + def test_load_raises_on_corrupted_json(self, tmp_path): + from specify_cli.workflows.step.catalog import StepRegistry + + steps_dir = tmp_path / ".specify" / "workflows" / "steps" + steps_dir.mkdir(parents=True) + registry_path = steps_dir / StepRegistry.REGISTRY_FILE + registry_path.write_text("not valid json {{{", encoding="utf-8") + + with pytest.raises(OSError, match="corrupted"): + StepRegistry(tmp_path) + + def test_save_uses_atomic_write(self, tmp_path): + from specify_cli.workflows.step.catalog import StepRegistry + + registry = StepRegistry(tmp_path) + registry.add("test-step", {"type": "command", "config": {}}) + registry.save() + + registry_path = ( + tmp_path / ".specify" / "workflows" / "steps" / StepRegistry.REGISTRY_FILE + ) + assert registry_path.exists() + data = json.loads(registry_path.read_text(encoding="utf-8")) + assert "test-step" in data.get("steps", {}) + assert not list(registry_path.parent.glob(".*.tmp"))