Skip to content

Commit 9ffaf6d

Browse files
authored
Harden step registry save and custom step/catalog ID handling
1 parent 2ad6627 commit 9ffaf6d

3 files changed

Lines changed: 69 additions & 2 deletions

File tree

src/specify_cli/workflows/__init__.py

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -109,7 +109,9 @@ def load_custom_steps(project_root: Path) -> list[str]:
109109
continue
110110
step_yml = step_dir / "step.yml"
111111
init_py = step_dir / "__init__.py"
112-
if not step_yml.exists() or not init_py.exists():
112+
if step_yml.is_symlink() or init_py.is_symlink():
113+
continue
114+
if not step_yml.is_file() or not init_py.is_file():
113115
continue
114116

115117
try:

src/specify_cli/workflows/catalog.py

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -677,6 +677,10 @@ def save(self) -> None:
677677
errors (read-only fs, permission denied, ...) so callers can surface
678678
a clean error to the user rather than an unhandled ``OSError``.
679679
"""
680+
if self._has_symlinked_parent() or self.registry_path.is_symlink():
681+
raise StepValidationError(
682+
"Refusing to write step registry through a symlinked path."
683+
)
680684
try:
681685
self.steps_dir.mkdir(parents=True, exist_ok=True)
682686
with open(self.registry_path, "w", encoding="utf-8") as f:
@@ -1000,8 +1004,10 @@ def _get_merged_steps(
10001004
for step_data in steps:
10011005
if not isinstance(step_data, dict):
10021006
continue
1003-
step_id = step_data.get("id", "")
1007+
raw_step_id = step_data.get("id", "")
1008+
step_id = str(raw_step_id or "").strip()
10041009
if step_id:
1010+
step_data["id"] = step_id
10051011
step_data["_catalog_name"] = entry.name
10061012
step_data["_install_allowed"] = entry.install_allowed
10071013
merged[step_id] = step_data

tests/test_workflows.py

Lines changed: 59 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2071,6 +2071,20 @@ def test_registry_load_refuses_symlinked_steps_dir(self, project_dir):
20712071
registry = StepRegistry(project_dir)
20722072
assert registry.list() == {}
20732073

2074+
@pytest.mark.skipif(not hasattr(os, "symlink"), reason="symlinks are unavailable")
2075+
def test_registry_save_refuses_symlinked_steps_dir(self, project_dir):
2076+
"""save() must refuse symlinked registry paths (defense-in-depth)."""
2077+
from specify_cli.workflows.catalog import StepRegistry, StepValidationError
2078+
2079+
outside = project_dir.parent / "outside-steps-save"
2080+
outside.mkdir(parents=True, exist_ok=True)
2081+
steps_link = project_dir / ".specify" / "workflows" / "steps"
2082+
steps_link.symlink_to(outside, target_is_directory=True)
2083+
2084+
registry = StepRegistry(project_dir)
2085+
with pytest.raises(StepValidationError, match="symlinked path"):
2086+
registry.save()
2087+
20742088

20752089
# ===== Step Catalog Tests =====
20762090

@@ -2296,6 +2310,31 @@ def test_search_with_non_string_fields(self, project_dir, monkeypatch):
22962310
results = catalog.search(query="missing")
22972311
assert len(results) == 0
22982312

2313+
def test_get_merged_steps_normalizes_list_ids_to_strings(self, project_dir, monkeypatch):
2314+
"""List-based catalog entries with non-string ids must be normalized."""
2315+
from specify_cli.workflows.catalog import StepCatalog, StepCatalogEntry
2316+
2317+
catalog = StepCatalog(project_dir)
2318+
entry = StepCatalogEntry(
2319+
name="test",
2320+
url="https://example.com/steps.json",
2321+
priority=1,
2322+
install_allowed=True,
2323+
)
2324+
monkeypatch.setattr(catalog, "get_active_catalogs", lambda: [entry])
2325+
monkeypatch.setattr(
2326+
catalog,
2327+
"_fetch_single_catalog",
2328+
lambda _entry, _force_refresh=False: {
2329+
"steps": [{"id": 42, "name": "Integer ID"}]
2330+
},
2331+
)
2332+
2333+
merged = catalog._get_merged_steps()
2334+
assert "42" in merged
2335+
assert 42 not in merged
2336+
assert merged["42"]["id"] == "42"
2337+
22992338
from specify_cli.workflows.catalog import StepCatalog
23002339

23012340
catalog = StepCatalog(project_dir)
@@ -2389,6 +2428,26 @@ def test_skip_missing_init_py(self, project_dir):
23892428
loaded = load_custom_steps(project_dir)
23902429
assert "bad-step2" not in loaded
23912430

2431+
@pytest.mark.skipif(not hasattr(os, "symlink"), reason="symlinks are unavailable")
2432+
def test_skip_symlinked_step_files(self, project_dir):
2433+
from specify_cli.workflows import load_custom_steps
2434+
2435+
step_dir = project_dir / ".specify" / "workflows" / "steps" / "bad-symlinked-files"
2436+
step_dir.mkdir(parents=True)
2437+
2438+
outside = project_dir.parent / "outside-step-files"
2439+
outside.mkdir(parents=True, exist_ok=True)
2440+
step_yml_target = outside / "step.yml"
2441+
step_yml_target.write_text("step:\n type_key: bad-symlinked-files\n", encoding="utf-8")
2442+
init_target = outside / "__init__.py"
2443+
init_target.write_text("# external code", encoding="utf-8")
2444+
2445+
(step_dir / "step.yml").symlink_to(step_yml_target)
2446+
(step_dir / "__init__.py").symlink_to(init_target)
2447+
2448+
loaded = load_custom_steps(project_dir)
2449+
assert "bad-symlinked-files" not in loaded
2450+
23922451
def test_skip_already_registered(self, project_dir):
23932452
from specify_cli.workflows import load_custom_steps
23942453

0 commit comments

Comments
 (0)