From b52b1ec70f967ff74453b649e0b06067399fe3a2 Mon Sep 17 00:00:00 2001 From: Quratulain-bilal Date: Fri, 31 Jul 2026 20:47:44 +0500 Subject: [PATCH 1/4] fix: make StepRegistry atomic and raise on corruption instead of silent data loss 1. _load() now raises OSError on symlinked paths and corrupted files instead of silently returning empty defaults (matches WorkflowRegistry) 2. save() now writes to a temp file and atomically replaces, preventing data loss from truncated writes on crash --- src/specify_cli/workflows/catalog.py | 62 ++++++++++++++++++++-------- 1 file changed, 45 insertions(+), 17 deletions(-) diff --git a/src/specify_cli/workflows/catalog.py b/src/specify_cli/workflows/catalog.py index 1c7354203b..7b63eeec52 100644 --- a/src/specify_cli/workflows/catalog.py +++ b/src/specify_cli/workflows/catalog.py @@ -885,28 +885,43 @@ def _load(self) -> dict[str, Any]: # 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._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: - """Persist registry to disk. + """Persist registry to disk atomically. Raises ``StepValidationError`` with a clear message on filesystem errors (read-only fs, permission denied, ...) so callers can surface @@ -918,8 +933,21 @@ def save(self) -> None: ) 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) + fd, tmp = tempfile.mkstemp( + dir=str(self.registry_path.parent), + prefix=f".{self.registry_path.name}.", + suffix=".tmp", + ) + try: + with os.fdopen(fd, "w", encoding="utf-8") as f: + json.dump(self.data, f, indent=2) + os.replace(tmp, self.registry_path) + except BaseException: + try: + os.unlink(tmp) + except OSError: + pass + raise except OSError as exc: raise StepValidationError( f"Failed to write step registry at {self.registry_path}: {exc}" From 64146875d8063980489f5efa92b7ce95fe80b9c1 Mon Sep 17 00:00:00 2001 From: Quratulain-bilal Date: Wed, 2 Sep 2026 00:33:39 +0500 Subject: [PATCH 2/4] test: add regression tests for StepRegistry atomic save Verifies: - StepRegistry raises OSError on symlinked registry files - StepRegistry raises OSError on corrupted JSON - StepRegistry.save() uses atomic write via mkstemp + os.replace Addresses Copilot review feedback on #3922. --- tests/test_workflows.py | 45 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 45 insertions(+) diff --git a/tests/test_workflows.py b/tests/test_workflows.py index 8a1bdbf38e..72d192b8df 100644 --- a/tests/test_workflows.py +++ b/tests/test_workflows.py @@ -16447,3 +16447,48 @@ def test_enable_disable_idempotent_warnings(self, project_dir, monkeypatch): result = runner.invoke(app, ["workflow", "disable", "align-wf"]) assert result.exit_code == 0 assert "already disabled" in result.output + + +class TestStepRegistryAtomicSaveRegression: + """Regression: StepRegistry must use atomic save and raise on corruption.""" + + def test_load_raises_on_symlinked_file(self, tmp_path): + """StepRegistry must raise OSError on symlinked registry file.""" + import json + from specify_cli.workflows.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): + """StepRegistry must raise OSError on corrupted JSON.""" + from specify_cli.workflows.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): + """StepRegistry.save() must use atomic write via mkstemp + os.replace.""" + from specify_cli.workflows.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() + import json + data = json.loads(registry_path.read_text(encoding="utf-8")) + assert "test-step" in data.get("steps", {}) From 86b69d4a0b0d34bb80e1a2ec101fcc1155b03031 Mon Sep 17 00:00:00 2001 From: Quratulain-bilal Date: Tue, 6 Oct 2026 21:45:12 +0500 Subject: [PATCH 3/4] fix(workflows): raise on corrupt step registry, save atomically Port the branch's StepRegistry hardening to where main now keeps the class (workflows/step/catalog/_domain.py): _load raises OSError for a symlinked path, an unreadable file, unparseable JSON, or a non-registry shape instead of silently returning an empty registry, and save writes through a temp file plus os.replace so a crash cannot leave a half-written registry. The new constructor errors would otherwise surface as tracebacks, and the step commands used to reach their directory-symlink check only because the registry swallowed the fault. A shared _load_step_registry_or_exit helper now prints a clean error, and add/remove resolve the steps base directory first so the symlinked-directory message keeps its specific wording. Five tests that pinned the silent-reset behaviour are updated to the fail-loud contract, and the branch's three regression tests (symlink, corrupted JSON, atomic save) come along. Assisted-by: opencode (model: mimo-v2.6-flash-free, autonomous) --- src/specify_cli/workflows/step/_helpers.py | 16 ++++ .../workflows/step/catalog/_domain.py | 68 ++++++++++----- src/specify_cli/workflows/step/command_add.py | 9 +- .../workflows/step/command_info.py | 5 +- .../workflows/step/command_list.py | 6 +- .../workflows/step/command_remove.py | 8 +- tests/test_workflows.py | 84 +++++++++++++++---- 7 files changed, 147 insertions(+), 49 deletions(-) diff --git a/src/specify_cli/workflows/step/_helpers.py b/src/specify_cli/workflows/step/_helpers.py index 45250ceffc..820cac3529 100644 --- a/src/specify_cli/workflows/step/_helpers.py +++ b/src/specify_cli/workflows/step/_helpers.py @@ -105,3 +105,19 @@ def _resolve_steps_base_dir_or_exit(project_root: cli.Path) -> cli.Path: raise cli.typer.Exit(1) return steps_base_dir + + +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] {exc}") + raise cli.typer.Exit(1) + diff --git a/src/specify_cli/workflows/step/catalog/_domain.py b/src/specify_cli/workflows/step/catalog/_domain.py index 08a1b22f57..7d3caa7021 100644 --- a/src/specify_cli/workflows/step/catalog/_domain.py +++ b/src/specify_cli/workflows/step/catalog/_domain.py @@ -5,6 +5,7 @@ import hashlib import json import os +import tempfile import time from dataclasses import dataclass from pathlib import Path @@ -85,33 +86,49 @@ 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: - """Persist registry to disk. + """Persist registry to disk atomically. Raises ``StepValidationError`` with a clear message on filesystem errors (read-only fs, permission denied, ...) so callers can surface @@ -123,8 +140,21 @@ def save(self) -> None: ) 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) + fd, tmp = tempfile.mkstemp( + dir=str(self.registry_path.parent), + prefix=f".{self.registry_path.name}.", + suffix=".tmp", + ) + try: + with os.fdopen(fd, "w", encoding="utf-8") as f: + json.dump(self.data, f, indent=2) + os.replace(tmp, self.registry_path) + except BaseException: + try: + os.unlink(tmp) + except OSError: + pass + raise except OSError as exc: raise StepValidationError( f"Failed to write step registry at {self.registry_path}: {exc}" diff --git a/src/specify_cli/workflows/step/command_add.py b/src/specify_cli/workflows/step/command_add.py index 7a3a3b8cad..1984d76200 100644 --- a/src/specify_cli/workflows/step/command_add.py +++ b/src/specify_cli/workflows/step/command_add.py @@ -16,7 +16,6 @@ def workflow_step_add( from .catalog import ( StepCatalog, StepCatalogError, - StepRegistry, StepValidationError, ) @@ -51,8 +50,10 @@ def workflow_step_add( ) raise cli.typer.Exit(1) - # Reject if already installed - registry = StepRegistry(project_root) + # Directory-level symlink refusal runs first so it keeps its specific + # message; registry construction then surfaces read errors cleanly. + step_helpers._resolve_steps_base_dir_or_exit(project_root) + registry = step_helpers._load_step_registry_or_exit(project_root) if registry.is_installed(step_id): cli.console.print( f"[red]Error:[/red] Step type '{step_id}' is already installed. " @@ -346,7 +347,7 @@ def _safe_fetch(url: str) -> bytes: step_version = info.get("version") or step_meta.get("version") or "0.0.0" # Register in step registry - registry = StepRegistry(project_root) + registry = step_helpers._load_step_registry_or_exit(project_root) try: registry.add( step_id, diff --git a/src/specify_cli/workflows/step/command_info.py b/src/specify_cli/workflows/step/command_info.py index c98072d8c0..df0c177715 100644 --- a/src/specify_cli/workflows/step/command_info.py +++ b/src/specify_cli/workflows/step/command_info.py @@ -12,12 +12,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 58ec7602f3..2eec8dfd44 100644 --- a/src/specify_cli/workflows/step/command_remove.py +++ b/src/specify_cli/workflows/step/command_remove.py @@ -13,16 +13,18 @@ def workflow_step_remove( step_id: str = cli.typer.Argument(..., help="Step type ID to uninstall"), ): """Uninstall a custom step type.""" - from .catalog import StepRegistry, StepValidationError + from .catalog import StepValidationError project_root = cli._require_specify_project() step_helpers._validate_step_id_or_exit(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/tests/test_workflows.py b/tests/test_workflows.py index acfdaa0b3b..9839023ce9 100644 --- a/tests/test_workflows.py +++ b/tests/test_workflows.py @@ -9325,19 +9325,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 @@ -9349,15 +9350,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 @@ -9371,12 +9369,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") @@ -9395,20 +9394,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() @@ -10773,3 +10775,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")) From 466d635a6ba7c29f69c823307cce942d13949a1b Mon Sep 17 00:00:00 2001 From: Quratulain-bilal Date: Tue, 6 Oct 2026 22:03:59 +0500 Subject: [PATCH 4/4] fix(workflows): raise on corrupt step registry StepRegistry._load returned an empty registry for a symlinked path, an unreadable file, unparseable JSON, or a non-registry shape, so corruption reached callers as a silently reset registry. It now raises OSError with the registry path in the message; save() already writes atomically on main. Because the constructor now raises, translate the error where it surfaces: installer._load_step_registry maps OSError onto StepInstallError so the existing command handlers keep printing a clean error, and the new _helpers._load_step_registry_or_exit does the same for the step commands. workflow_step_remove resolves the steps base directory before loading the registry so the symlinked-directory message keeps its specific wording. Tests: the three fail-loud regression tests (symlinked file, corrupted JSON, atomic save) are kept, and five tests that pinned the silent-reset behaviour now assert the raised OSError instead. Assisted-by: opencode (model: mimo-v2.6-flash-free, autonomous) --- src/specify_cli/workflows/step/_helpers.py | 15 ++++++ .../workflows/step/catalog/_domain.py | 48 ++++++++++++------- .../workflows/step/command_remove.py | 8 ++-- src/specify_cli/workflows/step/installer.py | 25 +++++++--- 4 files changed, 70 insertions(+), 26 deletions(-) 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_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,