fix(workflows): guard directory-shaped workflow.yml and unreadable registry

- workflow add's plain local-path fallback (no --dev) checked wf_file.exists()
  before installing, so a directory literally named workflow.yml passed the
  guard and _validate_and_install_local() leaked an uncaught
  IsADirectoryError instead of the documented CLI error. Use is_file(),
  matching the --dev branch's existing guard.
- WorkflowRegistry._load() treated any OSError while reading an existing
  registry the same as corrupted JSON, resetting to an empty in-memory
  registry. A later save() would then silently persist that empty state via
  os.replace, discarding every previously installed workflow entry. Track a
  _load_error flag on OSError-during-read and have save() refuse to write
  when it is set, so a transient I/O failure can no longer overwrite intact
  data on disk.
- docs/reference/workflows.md: document `--from <url>` with its value
  placeholder, matching extensions.md and presets.md.

Assisted-by: GitHub Copilot (model: Claude Sonnet 5, autonomous)
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
marcelsafin
2026-07-11 01:38:47 +02:00
parent 6a0c462f69
commit 3763e6fa35
4 changed files with 72 additions and 12 deletions

View File

@@ -86,10 +86,10 @@ Lists workflows installed in the current project.
specify workflow add <source>
```
| Option | Description |
| -------- | ------------------------------------------------------ |
| `--dev` | Install from a local workflow YAML file or directory |
| `--from` | Install from a custom URL (`<source>` names the expected workflow ID) |
| Option | Description |
| --------------- | ------------------------------------------------------ |
| `--dev` | Install from a local workflow YAML file or directory |
| `--from <url>` | Install from a custom URL (`<source>` names the expected workflow ID) |
Installs a workflow from the catalog, a URL (HTTPS required), or a local file path.

View File

@@ -825,7 +825,7 @@ def workflow_add(
return
elif source_path.is_dir():
wf_file = source_path / "workflow.yml"
if not wf_file.exists():
if not wf_file.is_file():
console.print(f"[red]Error:[/red] No workflow.yml found in {_escape_markup(source)}")
raise typer.Exit(1)
_validate_and_install_local(wf_file, str(source_path))

View File

@@ -70,6 +70,10 @@ class WorkflowRegistry:
self.project_root = project_root
self.workflows_dir = project_root / ".specify" / "workflows"
self.registry_path = self.workflows_dir / self.REGISTRY_FILE
# Set before _load() so a read failure (distinct from a corrupted or
# missing file) can flip it to block a later save() from silently
# persisting an empty registry over unreadable-but-intact data.
self._load_error = False
self.data = self._load()
def _has_symlinked_parent(self) -> bool:
@@ -95,15 +99,23 @@ class WorkflowRegistry:
try:
with open(self.registry_path, encoding="utf-8") as f:
data = json.load(f)
# Validate shape: must be a dict with a dict "workflows" field.
if not isinstance(data, dict):
return default_registry
if not isinstance(data.get("workflows"), dict):
data["workflows"] = {}
return data
except (json.JSONDecodeError, ValueError, OSError, UnicodeError):
except OSError:
# An I/O failure (e.g. permissions, transient FS issue) is not
# the same as a corrupted file: the real data may still be
# intact on disk. Flag it so save() refuses to overwrite that
# data with this in-memory default instead of silently
# discarding every prior entry.
self._load_error = True
return default_registry
except (json.JSONDecodeError, ValueError, UnicodeError):
# Corrupted registry file — reset to default
return default_registry
# Validate shape: must be a dict with a dict "workflows" field.
if not isinstance(data, dict):
return default_registry
if not isinstance(data.get("workflows"), dict):
data["workflows"] = {}
return data
return default_registry
def save(self) -> None:
@@ -114,6 +126,12 @@ class WorkflowRegistry:
raise OSError(
"Refusing to write workflow registry through a symlinked path."
)
if self._load_error:
raise OSError(
f"Refusing to save workflow registry at {self.registry_path}: "
"the existing file could not be read, so saving now would "
"discard its contents."
)
self.workflows_dir.mkdir(parents=True, exist_ok=True)
# Unique, exclusive temp then replace: a failed dump cannot truncate
# the registry, a pre-created symlink cannot redirect the write, and

View File

@@ -4746,6 +4746,33 @@ class TestWorkflowRegistry:
registry2 = WorkflowRegistry(project_dir)
assert registry2.is_installed("test-wf")
def test_load_read_oserror_refuses_to_save_over_existing_data(self, project_dir, monkeypatch):
"""A transient read failure (e.g. temporarily unreadable file) must not be
treated the same as a corrupted/missing registry: saving afterwards would
silently discard every previously persisted workflow entry."""
from specify_cli.workflows.catalog import WorkflowRegistry
import builtins
registry1 = WorkflowRegistry(project_dir)
registry1.add("test-wf", {"name": "Test", "version": "1.0.0"})
registry_path = registry1.registry_path
real_open = builtins.open
def _raising_open(file, mode="r", *args, **kwargs):
if Path(file) == registry_path and "r" in mode:
raise OSError("simulated read failure")
return real_open(file, mode, *args, **kwargs)
monkeypatch.setattr(builtins, "open", _raising_open)
registry2 = WorkflowRegistry(project_dir)
# The in-memory view may fall back to empty, but a save must not be
# allowed to persist that empty state over the real file on disk.
with pytest.raises(OSError):
registry2.save()
# The original entry must survive on disk untouched.
data = json.loads(registry_path.read_text(encoding="utf-8"))
assert "test-wf" in data["workflows"]
# ===== Workflow Catalog Tests =====
@@ -7314,6 +7341,21 @@ steps:
assert "No workflow.yml found" in result.output
assert "[bracket]" in result.output
def test_add_local_dir_with_workflow_yml_directory_errors_cleanly(self, project_dir, monkeypatch):
"""Same as the --dev case, but for the plain local-path fallback (no --dev):
a directory named workflow.yml must not reach open() and leak IsADirectoryError."""
from typer.testing import CliRunner
from specify_cli import app
monkeypatch.chdir(project_dir)
src_dir = project_dir / "local-wf"
(src_dir / "workflow.yml").mkdir(parents=True)
runner = CliRunner()
result = runner.invoke(app, ["workflow", "add", str(src_dir)])
assert result.exit_code != 0
assert result.exception is None or isinstance(result.exception, SystemExit)
assert "No workflow.yml found" in result.output
def test_add_yaml_parse_error_escapes_rich_markup(self, project_dir, monkeypatch):
"""A YAML syntax error can quote the offending line verbatim; brackets in it must not be Rich markup."""
from unittest.mock import patch