fix(workflows): reject catalog updates whose downloaded version mismatches

The update path never verified the downloaded workflow carries the catalog
version that triggered the update, so a stale or misconfigured URL could
report success while leaving the old version installed or downgrading it.
Pass the expected version into the install helper and fail the update when
the downloaded definition does not match.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
marcelsafin
2026-07-10 23:13:55 +02:00
parent a5307256ca
commit 560c7cdd5f
2 changed files with 74 additions and 1 deletions

View File

@@ -789,11 +789,14 @@ def _install_workflow_from_catalog(
registry: Any,
workflows_dir: Path,
workflow_id: str,
expected_version: str | None = None,
) -> None:
"""Download, validate, and register a catalog workflow.
Shared by ``workflow add`` and ``workflow update``. Raises ``typer.Exit``
on any failure; the registry entry is only written on full success.
``expected_version``, when given, rejects a downloaded workflow whose
version does not match the catalog version that triggered the install.
"""
from .catalog import WorkflowCatalog, WorkflowCatalogError
from .engine import WorkflowDefinition
@@ -928,6 +931,25 @@ def _install_workflow_from_catalog(
)
raise typer.Exit(1)
# A stale or misconfigured URL can serve a different version than the
# catalog advertised; without this check `update` would report success
# while leaving the old version installed (or even downgrading).
if expected_version is not None:
from packaging import version as pkg_version
try:
version_matches = pkg_version.Version(str(definition.version)) == pkg_version.Version(expected_version)
except pkg_version.InvalidVersion:
version_matches = str(definition.version) == expected_version
if not version_matches:
import shutil
shutil.rmtree(workflow_dir, ignore_errors=True)
console.print(
f"[red]Error:[/red] Downloaded workflow version ({_escape_markup(str(definition.version))}) "
f"does not match the catalog version ({_escape_markup(expected_version)}). "
f"The catalog entry may be stale or misconfigured."
)
raise typer.Exit(1)
entry = {
"name": definition.name or info.get("name", workflow_id),
"version": definition.version or info.get("version", "0.0.0"),
@@ -1109,7 +1131,10 @@ def workflow_update(
wf_dir = _safe_workflow_id_dir(workflows_dir, update["id"])
wf_file = wf_dir / "workflow.yml"
backup = wf_file.read_bytes() if wf_file.is_file() else None
_install_workflow_from_catalog(project_root, registry, workflows_dir, update["id"])
_install_workflow_from_catalog(
project_root, registry, workflows_dir, update["id"],
expected_version=update["available"],
)
except (typer.Exit, OSError) as exc:
if backup is not None and wf_dir is not None and wf_file is not None:
try:

View File

@@ -7660,6 +7660,54 @@ steps:
# The previously installed workflow must survive.
assert "1.0.0" in (wf_dir / "workflow.yml").read_text(encoding="utf-8")
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
from typer.testing import CliRunner
from specify_cli import app
from specify_cli.workflows.catalog import WorkflowCatalog, WorkflowRegistry
monkeypatch.chdir(project_dir)
WorkflowRegistry(project_dir).add("align-wf", {
"name": "Align Workflow",
"version": "1.0.0",
"description": "CLI alignment test workflow",
"source": "catalog",
"catalog_name": "test-catalog",
"url": "https://example.com/workflow.yml",
})
wf_dir = project_dir / ".specify" / "workflows" / "align-wf"
wf_dir.mkdir(parents=True)
(wf_dir / "workflow.yml").write_text(
self.WORKFLOW_YAML.format(version="1.0.0"), encoding="utf-8"
)
monkeypatch.setattr(
WorkflowCatalog,
"get_workflow_info",
lambda self, wid: {
"id": wid,
"name": "Align Workflow",
"version": "2.0.0",
"url": "https://example.com/workflow.yml",
"_install_allowed": True,
"_catalog_name": "test-catalog",
},
)
# The URL still serves the old 1.0.0 payload.
data = self.WORKFLOW_YAML.format(version="1.0.0").encode()
runner = CliRunner()
with patch(
"specify_cli.authentication.http.open_url",
side_effect=lambda url, timeout=None, extra_headers=None: self._FakeResponse(data, url),
):
result = runner.invoke(app, ["workflow", "update"], input="y\n")
assert "does not match the catalog version" in result.output
assert "Failed to update" in result.output
meta = WorkflowRegistry(project_dir).get("align-wf")
assert meta["version"] == "1.0.0"
assert "1.0.0" in (wf_dir / "workflow.yml").read_text(encoding="utf-8")
def test_update_preserves_disabled_state(self, project_dir, monkeypatch):
from unittest.mock import patch
from typer.testing import CliRunner