fix(workflows): guard non-string catalog URL and keep enable/disable rollback intact

- A truthy non-string catalog url (e.g. 123) reached urlparse and raised
  AttributeError, escaping the clean error path; validate it is a string.
- enable/disable mutated the live registry entry before add(), so add's
  rollback snapshot captured the already-toggled object; pass a fresh
  mapping instead.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
marcelsafin
2026-07-11 00:05:26 +02:00
parent f28e125338
commit 338f7eecb0
2 changed files with 60 additions and 4 deletions

View File

@@ -846,6 +846,12 @@ def _install_workflow_from_catalog(
if not workflow_url:
console.print(f"[red]Error:[/red] Workflow '{safe_wf_id}' does not have an install URL in the catalog")
raise typer.Exit(1)
if not isinstance(workflow_url, str):
# Untrusted catalog payload; a non-string would crash urlparse below.
console.print(
f"[red]Error:[/red] Workflow '{safe_wf_id}' has a malformed install URL."
)
raise typer.Exit(1)
# Validate URL scheme (HTTPS required, HTTP allowed for localhost only)
from ipaddress import ip_address
@@ -1209,8 +1215,9 @@ def workflow_enable(
if metadata.get("enabled", True):
console.print(f"[yellow]Workflow '{_escape_markup(workflow_id)}' is already enabled[/yellow]")
raise typer.Exit(0)
metadata["enabled"] = True
registry.add(workflow_id, metadata)
# Fresh mapping: registry.get() returns the live entry, and mutating it
# in place would defeat WorkflowRegistry.add's rollback-on-save-failure.
registry.add(workflow_id, {**metadata, "enabled": True})
console.print(f"[green]✓[/green] Workflow '{_escape_markup(workflow_id)}' enabled")
@@ -1235,8 +1242,8 @@ def workflow_disable(
if not metadata.get("enabled", True):
console.print(f"[yellow]Workflow '{_escape_markup(workflow_id)}' is already disabled[/yellow]")
raise typer.Exit(0)
metadata["enabled"] = False
registry.add(workflow_id, metadata)
# Fresh mapping for the same rollback reason as workflow_enable.
registry.add(workflow_id, {**metadata, "enabled": False})
console.print(f"[green]✓[/green] Workflow '{_escape_markup(workflow_id)}' disabled")
console.print(f"To re-enable: specify workflow enable {_escape_markup(workflow_id)}")

View File

@@ -7714,6 +7714,55 @@ steps:
# The previously installed workflow must survive.
assert "1.0.0" in (wf_dir / "workflow.yml").read_text(encoding="utf-8")
def test_add_non_string_catalog_url_fails_cleanly(self, project_dir, monkeypatch):
"""A truthy non-string catalog URL must hit the clean error path, not AttributeError."""
from typer.testing import CliRunner
from specify_cli import app
from specify_cli.workflows.catalog import WorkflowCatalog
monkeypatch.chdir(project_dir)
monkeypatch.setattr(
WorkflowCatalog,
"get_workflow_info",
lambda self, wid: {
"id": wid,
"name": "Align Workflow",
"version": "1.0.0",
"url": 123,
"_install_allowed": True,
"_catalog_name": "test-catalog",
},
)
runner = CliRunner()
result = runner.invoke(app, ["workflow", "add", "align-wf"])
assert result.exit_code != 0
assert result.exception is None or isinstance(result.exception, SystemExit)
assert "malformed install URL" in result.output
def test_enable_failed_save_leaves_workflow_disabled(self, project_dir, monkeypatch):
from typer.testing import CliRunner
from specify_cli import app
from specify_cli.workflows.catalog import WorkflowRegistry
monkeypatch.chdir(project_dir)
runner = CliRunner()
self._install_dev(runner, app, project_dir)
result = runner.invoke(app, ["workflow", "disable", "align-wf"])
assert result.exit_code == 0, result.output
def boom(self):
raise OSError("disk full")
with pytest.MonkeyPatch.context() as mp:
mp.setattr(WorkflowRegistry, "save", boom)
result = runner.invoke(app, ["workflow", "enable", "align-wf"])
assert result.exit_code != 0
assert WorkflowRegistry(project_dir).get("align-wf")["enabled"] is False
result = runner.invoke(app, ["workflow", "enable", "align-wf"])
assert result.exit_code == 0, result.output
assert WorkflowRegistry(project_dir).get("align-wf")["enabled"] is True
def test_update_rejects_version_mismatch_from_stale_url(self, project_dir, monkeypatch):
"""A URL serving a different version than the catalog advertised must fail the update."""
from unittest.mock import patch