From 338f7eecb0edaae22becedccd3273425e1bc9d13 Mon Sep 17 00:00:00 2001 From: marcelsafin <179933638+marcelsafin@users.noreply.github.com> Date: Sat, 11 Jul 2026 00:05:26 +0200 Subject: [PATCH] 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> --- src/specify_cli/workflows/_commands.py | 15 +++++--- tests/test_workflows.py | 49 ++++++++++++++++++++++++++ 2 files changed, 60 insertions(+), 4 deletions(-) diff --git a/src/specify_cli/workflows/_commands.py b/src/specify_cli/workflows/_commands.py index 691fa8945..6ce5ceb27 100644 --- a/src/specify_cli/workflows/_commands.py +++ b/src/specify_cli/workflows/_commands.py @@ -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)}") diff --git a/tests/test_workflows.py b/tests/test_workflows.py index 083e2638e..8514017c6 100644 --- a/tests/test_workflows.py +++ b/tests/test_workflows.py @@ -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