fix(workflows): escape rich markup in list output and catalog install errors, isolate update failures

workflow list now escapes id/name/version/description before printing,
matching how extensions render user-editable fields. The catalog install
helper computes safe_wf_id once and uses it for every early error path
plus the final failure message.

workflow update wraps _safe_workflow_id_dir and the backup read inside
the try/except typer.Exit block so an unsafe id in a corrupted registry
fails that one workflow and the rest continue.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
marcelsafin
2026-07-09 14:57:06 +02:00
parent 52ab7ffdf0
commit 4c356c413e
2 changed files with 88 additions and 13 deletions

View File

@@ -582,14 +582,17 @@ def workflow_list():
console.print("\n[bold cyan]Installed Workflows:[/bold cyan]\n")
for wf_id, wf_data in installed.items():
safe_id = _escape_markup(wf_id)
if not isinstance(wf_data, dict):
console.print(f" [yellow]Warning:[/yellow] Skipping corrupted registry entry '{wf_id}'.\n")
console.print(f" [yellow]Warning:[/yellow] Skipping corrupted registry entry '{safe_id}'.\n")
continue
marker = "" if wf_data.get("enabled", True) else " [red]\\[disabled][/red]"
console.print(f" [bold]{wf_data.get('name', wf_id)}[/bold] ({wf_id}) v{wf_data.get('version', '?')}{marker}")
name = _escape_markup(str(wf_data.get("name", wf_id)))
version = _escape_markup(str(wf_data.get("version", "?")))
console.print(f" [bold]{name}[/bold] ({safe_id}) v{version}{marker}")
desc = wf_data.get("description", "")
if desc:
console.print(f" {desc}")
console.print(f" {_escape_markup(str(desc))}")
console.print()
@@ -790,6 +793,8 @@ def _install_workflow_from_catalog(
from .catalog import WorkflowCatalog, WorkflowCatalogError
from .engine import WorkflowDefinition
safe_wf_id = _escape_markup(workflow_id)
catalog = WorkflowCatalog(project_root)
try:
info = catalog.get_workflow_info(workflow_id)
@@ -798,17 +803,17 @@ def _install_workflow_from_catalog(
raise typer.Exit(1)
if not info:
console.print(f"[red]Error:[/red] Workflow '{workflow_id}' not found in catalog")
console.print(f"[red]Error:[/red] Workflow '{safe_wf_id}' not found in catalog")
raise typer.Exit(1)
if not info.get("_install_allowed", True):
console.print(f"[yellow]Warning:[/yellow] Workflow '{workflow_id}' is from a discovery-only catalog")
console.print(f"[yellow]Warning:[/yellow] Workflow '{safe_wf_id}' is from a discovery-only catalog")
console.print("Direct installation is not enabled for this catalog source.")
raise typer.Exit(1)
workflow_url = info.get("url")
if not workflow_url:
console.print(f"[red]Error:[/red] Workflow '{workflow_id}' does not have an install URL in the catalog")
console.print(f"[red]Error:[/red] Workflow '{safe_wf_id}' does not have an install URL in the catalog")
raise typer.Exit(1)
# Validate URL scheme (HTTPS required, HTTP allowed for localhost only)
@@ -828,7 +833,7 @@ def _install_workflow_from_catalog(
pass
if parsed_url.scheme != "https" and not (parsed_url.scheme == "http" and is_loopback):
console.print(
f"[red]Error:[/red] Workflow '{workflow_id}' has an invalid install URL. "
f"[red]Error:[/red] Workflow '{safe_wf_id}' has an invalid install URL. "
"Only HTTPS URLs are allowed, except HTTP for localhost/loopback."
)
raise typer.Exit(1)
@@ -869,7 +874,7 @@ def _install_workflow_from_catalog(
import shutil
shutil.rmtree(workflow_dir, ignore_errors=True)
console.print(
f"[red]Error:[/red] Workflow '{workflow_id}' redirected to non-HTTPS URL: {final_url}"
f"[red]Error:[/red] Workflow '{safe_wf_id}' redirected to non-HTTPS URL: {_escape_markup(final_url)}"
)
raise typer.Exit(1)
workflow_file.write_bytes(response.read())
@@ -879,7 +884,7 @@ def _install_workflow_from_catalog(
if workflow_dir.exists():
import shutil
shutil.rmtree(workflow_dir, ignore_errors=True)
console.print(f"[red]Error:[/red] Failed to install workflow '{workflow_id}' from catalog: {exc}")
console.print(f"[red]Error:[/red] Failed to install workflow '{safe_wf_id}' from catalog: {_escape_markup(str(exc))}")
raise typer.Exit(1)
# Validate the downloaded workflow before registering
@@ -1083,13 +1088,16 @@ def workflow_update(
for update in updates_available:
# Installed workflows are a single workflow.yml — back it up so a
# failed download/validation doesn't destroy the working copy.
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
wf_dir: Path | None = None
wf_file: Path | None = None
backup: bytes | None = None
try:
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"])
except typer.Exit:
if backup is not None:
if backup is not None and wf_dir is not None and wf_file is not None:
wf_dir.mkdir(parents=True, exist_ok=True)
wf_file.write_bytes(backup)
failed.append(update["id"])

View File

@@ -7541,6 +7541,73 @@ steps:
assert "corrupted" in result.output
assert "OK Workflow" in result.output
def test_list_escapes_rich_markup_in_registry_fields(self, project_dir, monkeypatch):
"""User-editable name/description/id fields must not be parsed as Rich markup."""
import json
from typer.testing import CliRunner
from specify_cli import app
from specify_cli.workflows.catalog import WorkflowRegistry
monkeypatch.chdir(project_dir)
registry_path = WorkflowRegistry(project_dir).registry_path
registry_path.parent.mkdir(parents=True, exist_ok=True)
registry_path.write_text(
json.dumps(
{
"schema_version": "1.0",
"workflows": {
"ok": {
"name": "Bracket [Test]",
"version": "1.0.0",
"description": "desc [with] brackets",
},
},
}
),
encoding="utf-8",
)
runner = CliRunner()
result = runner.invoke(app, ["workflow", "list"])
assert result.exit_code == 0, result.output
assert "Bracket [Test]" in result.output
assert "desc [with] brackets" in result.output
def test_update_reports_unsafe_registry_id_per_workflow(self, project_dir, monkeypatch):
"""An unsafe workflow id in the registry must fail that one entry, not abort the whole update."""
import json
from typer.testing import CliRunner
from specify_cli import app
from specify_cli.workflows.catalog import WorkflowRegistry, WorkflowCatalog
monkeypatch.chdir(project_dir)
registry_path = WorkflowRegistry(project_dir).registry_path
registry_path.parent.mkdir(parents=True, exist_ok=True)
registry_path.write_text(
json.dumps(
{
"schema_version": "1.0",
"workflows": {
"../evil": {
"name": "Bad",
"version": "0.0.1",
"source": "catalog",
"url": "https://example.com/evil.yml",
},
},
}
),
encoding="utf-8",
)
monkeypatch.setattr(
WorkflowCatalog,
"get_workflow_info",
lambda self, wid: {"version": "9.9.9", "url": "https://example.com/evil.yml", "_install_allowed": True},
)
runner = CliRunner()
result = runner.invoke(app, ["workflow", "update"], input="y\n")
assert result.exit_code != 0
assert "Failed to update" in result.output
def test_enable_disable_corrupted_registry_entry_errors(self, project_dir, monkeypatch):
import json
from typer.testing import CliRunner