Skip to content

Commit 3225b4e

Browse files
marcelsafinCopilot
andcommitted
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>
1 parent 4fcfd1a commit 3225b4e

2 files changed

Lines changed: 88 additions & 13 deletions

File tree

src/specify_cli/workflows/_commands.py

Lines changed: 21 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -582,14 +582,17 @@ def workflow_list():
582582

583583
console.print("\n[bold cyan]Installed Workflows:[/bold cyan]\n")
584584
for wf_id, wf_data in installed.items():
585+
safe_id = _escape_markup(wf_id)
585586
if not isinstance(wf_data, dict):
586-
console.print(f" [yellow]Warning:[/yellow] Skipping corrupted registry entry '{wf_id}'.\n")
587+
console.print(f" [yellow]Warning:[/yellow] Skipping corrupted registry entry '{safe_id}'.\n")
587588
continue
588589
marker = "" if wf_data.get("enabled", True) else " [red]\\[disabled][/red]"
589-
console.print(f" [bold]{wf_data.get('name', wf_id)}[/bold] ({wf_id}) v{wf_data.get('version', '?')}{marker}")
590+
name = _escape_markup(str(wf_data.get("name", wf_id)))
591+
version = _escape_markup(str(wf_data.get("version", "?")))
592+
console.print(f" [bold]{name}[/bold] ({safe_id}) v{version}{marker}")
590593
desc = wf_data.get("description", "")
591594
if desc:
592-
console.print(f" {desc}")
595+
console.print(f" {_escape_markup(str(desc))}")
593596
console.print()
594597

595598

@@ -782,6 +785,8 @@ def _install_workflow_from_catalog(
782785
from .catalog import WorkflowCatalog, WorkflowCatalogError
783786
from .engine import WorkflowDefinition
784787

788+
safe_wf_id = _escape_markup(workflow_id)
789+
785790
catalog = WorkflowCatalog(project_root)
786791
try:
787792
info = catalog.get_workflow_info(workflow_id)
@@ -790,17 +795,17 @@ def _install_workflow_from_catalog(
790795
raise typer.Exit(1)
791796

792797
if not info:
793-
console.print(f"[red]Error:[/red] Workflow '{workflow_id}' not found in catalog")
798+
console.print(f"[red]Error:[/red] Workflow '{safe_wf_id}' not found in catalog")
794799
raise typer.Exit(1)
795800

796801
if not info.get("_install_allowed", True):
797-
console.print(f"[yellow]Warning:[/yellow] Workflow '{workflow_id}' is from a discovery-only catalog")
802+
console.print(f"[yellow]Warning:[/yellow] Workflow '{safe_wf_id}' is from a discovery-only catalog")
798803
console.print("Direct installation is not enabled for this catalog source.")
799804
raise typer.Exit(1)
800805

801806
workflow_url = info.get("url")
802807
if not workflow_url:
803-
console.print(f"[red]Error:[/red] Workflow '{workflow_id}' does not have an install URL in the catalog")
808+
console.print(f"[red]Error:[/red] Workflow '{safe_wf_id}' does not have an install URL in the catalog")
804809
raise typer.Exit(1)
805810

806811
# Validate URL scheme (HTTPS required, HTTP allowed for localhost only)
@@ -820,7 +825,7 @@ def _install_workflow_from_catalog(
820825
pass
821826
if parsed_url.scheme != "https" and not (parsed_url.scheme == "http" and is_loopback):
822827
console.print(
823-
f"[red]Error:[/red] Workflow '{workflow_id}' has an invalid install URL. "
828+
f"[red]Error:[/red] Workflow '{safe_wf_id}' has an invalid install URL. "
824829
"Only HTTPS URLs are allowed, except HTTP for localhost/loopback."
825830
)
826831
raise typer.Exit(1)
@@ -861,7 +866,7 @@ def _install_workflow_from_catalog(
861866
import shutil
862867
shutil.rmtree(workflow_dir, ignore_errors=True)
863868
console.print(
864-
f"[red]Error:[/red] Workflow '{workflow_id}' redirected to non-HTTPS URL: {final_url}"
869+
f"[red]Error:[/red] Workflow '{safe_wf_id}' redirected to non-HTTPS URL: {_escape_markup(final_url)}"
865870
)
866871
raise typer.Exit(1)
867872
workflow_file.write_bytes(response.read())
@@ -871,7 +876,7 @@ def _install_workflow_from_catalog(
871876
if workflow_dir.exists():
872877
import shutil
873878
shutil.rmtree(workflow_dir, ignore_errors=True)
874-
console.print(f"[red]Error:[/red] Failed to install workflow '{workflow_id}' from catalog: {exc}")
879+
console.print(f"[red]Error:[/red] Failed to install workflow '{safe_wf_id}' from catalog: {_escape_markup(str(exc))}")
875880
raise typer.Exit(1)
876881

877882
# Validate the downloaded workflow before registering
@@ -1075,13 +1080,16 @@ def workflow_update(
10751080
for update in updates_available:
10761081
# Installed workflows are a single workflow.yml — back it up so a
10771082
# failed download/validation doesn't destroy the working copy.
1078-
wf_dir = _safe_workflow_id_dir(workflows_dir, update["id"])
1079-
wf_file = wf_dir / "workflow.yml"
1080-
backup = wf_file.read_bytes() if wf_file.is_file() else None
1083+
wf_dir: Path | None = None
1084+
wf_file: Path | None = None
1085+
backup: bytes | None = None
10811086
try:
1087+
wf_dir = _safe_workflow_id_dir(workflows_dir, update["id"])
1088+
wf_file = wf_dir / "workflow.yml"
1089+
backup = wf_file.read_bytes() if wf_file.is_file() else None
10821090
_install_workflow_from_catalog(project_root, registry, workflows_dir, update["id"])
10831091
except typer.Exit:
1084-
if backup is not None:
1092+
if backup is not None and wf_dir is not None and wf_file is not None:
10851093
wf_dir.mkdir(parents=True, exist_ok=True)
10861094
wf_file.write_bytes(backup)
10871095
failed.append(update["id"])

tests/test_workflows.py

Lines changed: 67 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7229,6 +7229,73 @@ def test_list_skips_corrupted_registry_entry(self, project_dir, monkeypatch):
72297229
assert "corrupted" in result.output
72307230
assert "OK Workflow" in result.output
72317231

7232+
def test_list_escapes_rich_markup_in_registry_fields(self, project_dir, monkeypatch):
7233+
"""User-editable name/description/id fields must not be parsed as Rich markup."""
7234+
import json
7235+
from typer.testing import CliRunner
7236+
from specify_cli import app
7237+
from specify_cli.workflows.catalog import WorkflowRegistry
7238+
7239+
monkeypatch.chdir(project_dir)
7240+
registry_path = WorkflowRegistry(project_dir).registry_path
7241+
registry_path.parent.mkdir(parents=True, exist_ok=True)
7242+
registry_path.write_text(
7243+
json.dumps(
7244+
{
7245+
"schema_version": "1.0",
7246+
"workflows": {
7247+
"ok": {
7248+
"name": "Bracket [Test]",
7249+
"version": "1.0.0",
7250+
"description": "desc [with] brackets",
7251+
},
7252+
},
7253+
}
7254+
),
7255+
encoding="utf-8",
7256+
)
7257+
runner = CliRunner()
7258+
result = runner.invoke(app, ["workflow", "list"])
7259+
assert result.exit_code == 0, result.output
7260+
assert "Bracket [Test]" in result.output
7261+
assert "desc [with] brackets" in result.output
7262+
7263+
def test_update_reports_unsafe_registry_id_per_workflow(self, project_dir, monkeypatch):
7264+
"""An unsafe workflow id in the registry must fail that one entry, not abort the whole update."""
7265+
import json
7266+
from typer.testing import CliRunner
7267+
from specify_cli import app
7268+
from specify_cli.workflows.catalog import WorkflowRegistry, WorkflowCatalog
7269+
7270+
monkeypatch.chdir(project_dir)
7271+
registry_path = WorkflowRegistry(project_dir).registry_path
7272+
registry_path.parent.mkdir(parents=True, exist_ok=True)
7273+
registry_path.write_text(
7274+
json.dumps(
7275+
{
7276+
"schema_version": "1.0",
7277+
"workflows": {
7278+
"../evil": {
7279+
"name": "Bad",
7280+
"version": "0.0.1",
7281+
"source": "catalog",
7282+
"url": "https://example.com/evil.yml",
7283+
},
7284+
},
7285+
}
7286+
),
7287+
encoding="utf-8",
7288+
)
7289+
monkeypatch.setattr(
7290+
WorkflowCatalog,
7291+
"get_workflow_info",
7292+
lambda self, wid: {"version": "9.9.9", "url": "https://example.com/evil.yml", "_install_allowed": True},
7293+
)
7294+
runner = CliRunner()
7295+
result = runner.invoke(app, ["workflow", "update"], input="y\n")
7296+
assert result.exit_code != 0
7297+
assert "Failed to update" in result.output
7298+
72327299
def test_enable_disable_corrupted_registry_entry_errors(self, project_dir, monkeypatch):
72337300
import json
72347301
from typer.testing import CliRunner

0 commit comments

Comments
 (0)