fix(integrations): escape catalog metadata in discovery output (#3772)

* fix(integrations): escape catalog metadata in discovery output Assisted-by: GitHub Copilot (model: GPT-5.6 Sol, autonomous) Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * fix(integrations): escape unknown query IDs Assisted-by: GitHub Copilot (model: GPT-5.6 Sol, autonomous) Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --------- Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
2026-08-03 06:26:30 +08:00 · 2026-07-28 17:20:31 +02:00
parent 999f8e6497
commit 2e44ed60e8
2 changed files with 103 additions and 21 deletions
--- a/src/specify_cli/integrations/_query_commands.py
+++ b/src/specify_cli/integrations/_query_commands.py
@@ -318,22 +318,26 @@ def integration_search(

    console.print(f"\n[green]Found {len(results)} integration(s):[/green]\n")
    for integ in sorted(results, key=lambda e: e.get("id", "")):
-        iid = integ.get("id", "?")
-        name = integ.get("name", iid)
-        version = integ.get("version", "?")
+        iid_value = str(integ.get("id", "?"))
+        iid = _rich_escape(iid_value)
+        name = _rich_escape(str(integ.get("name", iid_value)))
+        version = _rich_escape(str(integ.get("version", "?")))
        console.print(f"[bold]{name}[/bold] ({iid}) v{version}")
        desc = integ.get("description", "")
        if desc:
-            console.print(f"  {desc}")
+            console.print(f"  {_rich_escape(str(desc))}")

-        console.print(f"\n  [dim]Author:[/dim] {integ.get('author', 'Unknown')}")
+        author_value = _rich_escape(str(integ.get("author", "Unknown")))
+        console.print(f"\n  [dim]Author:[/dim] {author_value}")
        tags = integ.get("tags", [])
        if isinstance(tags, list) and tags:
-            console.print(f"  [dim]Tags:[/dim] {', '.join(str(t) for t in tags)}")
+            safe_tags = _rich_escape(", ".join(str(t) for t in tags))
+            console.print(f"  [dim]Tags:[/dim] {safe_tags}")

-        cat_name = integ.get("_catalog_name", "")
+        cat_name_value = integ.get("_catalog_name", "")
+        cat_name = _rich_escape(str(cat_name_value))
        install_allowed = integ.get("_install_allowed", True)
-        if cat_name:
+        if cat_name_value:
            if install_allowed:
                console.print(f"  [dim]Catalog:[/dim] {cat_name}")
            else:
@@ -342,9 +346,9 @@ def integration_search(
                    "[yellow](discovery only — not installable)[/yellow]"
                )

-        if iid == installed_key:
+        if iid_value == installed_key:
            console.print("\n  [green]✓ Installed[/green] (currently active)")
-        elif iid in INTEGRATION_REGISTRY:
+        elif iid_value in INTEGRATION_REGISTRY:
            console.print(f"\n  [cyan]Install:[/cyan] specify integration install {iid}")
        elif install_allowed:
            console.print(
@@ -374,6 +378,7 @@ def integration_info(
    project_root = _require_specify_project()
    catalog = IntegrationCatalog(project_root)
    installed_key = _default_integration_key(_read_integration_json(project_root))
+    safe_integration_id = _rich_escape(str(integration_id))

    try:
        info = catalog.get_integration_info(integration_id)
@@ -386,29 +391,38 @@ def integration_info(
        catalog_error = None

    if info:
-        name = info.get("name", integration_id)
-        version = info.get("version", "?")
-        console.print(f"\n[bold cyan]{name}[/bold cyan] ({integration_id}) v{version}")
+        name = _rich_escape(str(info.get("name", integration_id)))
+        version = _rich_escape(str(info.get("version", "?")))
+        console.print(
+            f"\n[bold cyan]{name}[/bold cyan] ({safe_integration_id}) v{version}"
+        )
        if info.get("description"):
-            console.print(f"  {info['description']}")
+            console.print(f"  {_rich_escape(str(info['description']))}")
        console.print()

-        console.print(f"  [dim]Author:[/dim] {info.get('author', 'Unknown')}")
+        author_value = _rich_escape(str(info.get("author", "Unknown")))
+        console.print(f"  [dim]Author:[/dim] {author_value}")
        if info.get("license"):
-            console.print(f"  [dim]License:[/dim] {info['license']}")
+            console.print(
+                f"  [dim]License:[/dim] {_rich_escape(str(info['license']))}"
+            )

        tags = info.get("tags", [])
        if isinstance(tags, list) and tags:
-            console.print(f"  [dim]Tags:[/dim] {', '.join(str(t) for t in tags)}")
+            safe_tags = _rich_escape(", ".join(str(t) for t in tags))
+            console.print(f"  [dim]Tags:[/dim] {safe_tags}")

-        cat_name = info.get("_catalog_name", "")
+        cat_name_value = info.get("_catalog_name", "")
+        cat_name = _rich_escape(str(cat_name_value))
        install_allowed = info.get("_install_allowed", True)
-        if cat_name:
+        if cat_name_value:
            install_note = "" if install_allowed else " [yellow](discovery only)[/yellow]"
            console.print(f"  [dim]Source catalog:[/dim] {cat_name}{install_note}")

        if info.get("repository"):
-            console.print(f"  [dim]Repository:[/dim] {info['repository']}")
+            console.print(
+                f"  [dim]Repository:[/dim] {_rich_escape(str(info['repository']))}"
+            )

        if integration_id == installed_key:
            console.print("\n  [green]✓ Installed[/green] (currently active)")
@@ -444,7 +458,7 @@ def integration_info(
        else:
            console.print("\nTry again when online, or use a built-in integration ID directly.")
    else:
-        console.print(f"[red]Error:[/red] Integration '{integration_id}' not found")
+        console.print(f"[red]Error:[/red] Integration '{safe_integration_id}' not found")
        console.print("\nTry: specify integration search")
    raise typer.Exit(1)

--- a/tests/integrations/test_cli.py
+++ b/tests/integrations/test_cli.py
@@ -1403,6 +1403,18 @@ class TestIntegrationCatalogDiscoveryCLI:
            "_install_allowed": True,
        },
    ]
+    MARKUP_INTEGRATION = {
+        "id": "[red]markup-id[/red]",
+        "name": "[green]Markup Name[/green]",
+        "version": "[blue]1.0.0[/blue]",
+        "description": "[yellow]Markup Description[/yellow]",
+        "author": "[magenta]Markup Author[/magenta]",
+        "license": "[cyan]Markup License[/cyan]",
+        "repository": "[bold]Markup Repository[/bold]",
+        "tags": ["[italic]markup-tag[/italic]"],
+        "_catalog_name": "[underline]markup-catalog[/underline]",
+        "_install_allowed": False,
+    }

    def _make_project(self, tmp_path):
        project = tmp_path / "proj"
@@ -1866,6 +1878,25 @@ class TestIntegrationCatalogDiscoveryCLI:
        # acme-coder is flagged _install_allowed=False, so we should warn
        assert "Not directly installable" in result.output

+    def test_search_escapes_catalog_markup(self, tmp_path, monkeypatch):
+        project = self._make_project(tmp_path)
+        self._patch_catalog(monkeypatch, integrations=[self.MARKUP_INTEGRATION])
+
+        result = self._invoke(["integration", "search"], project)
+
+        assert result.exit_code == 0, result.output
+        output = _normalize_cli_output(result.output)
+        for value in (
+            self.MARKUP_INTEGRATION["id"],
+            self.MARKUP_INTEGRATION["name"],
+            self.MARKUP_INTEGRATION["version"],
+            self.MARKUP_INTEGRATION["description"],
+            self.MARKUP_INTEGRATION["author"],
+            self.MARKUP_INTEGRATION["tags"][0],
+            self.MARKUP_INTEGRATION["_catalog_name"],
+        ):
+            assert value in output
+
    # -- info --------------------------------------------------------------

    def test_info_found(self, tmp_path, monkeypatch):
@@ -1888,6 +1919,19 @@ class TestIntegrationCatalogDiscoveryCLI:
        assert result.exit_code == 1
        assert "not found" in result.output

+    def test_info_not_found_escapes_query_markup(self, tmp_path, monkeypatch):
+        project = self._make_project(tmp_path)
+        self._patch_catalog(monkeypatch)
+        integration_id = "[red]does-not-exist[/red]"
+
+        result = self._invoke(
+            ["integration", "info", integration_id],
+            project,
+        )
+
+        assert result.exit_code == 1
+        assert integration_id in _normalize_cli_output(result.output)
+
    def test_info_builtin_not_in_catalog(self, tmp_path, monkeypatch):
        project = self._make_project(tmp_path)
        # Empty catalog, but copilot is a registered built-in.
@@ -1896,6 +1940,30 @@ class TestIntegrationCatalogDiscoveryCLI:
        assert result.exit_code == 0, result.output
        assert "Built-in integration" in result.output

+    def test_info_escapes_catalog_markup(self, tmp_path, monkeypatch):
+        project = self._make_project(tmp_path)
+        self._patch_catalog(monkeypatch, integrations=[self.MARKUP_INTEGRATION])
+
+        result = self._invoke(
+            ["integration", "info", self.MARKUP_INTEGRATION["id"]],
+            project,
+        )
+
+        assert result.exit_code == 0, result.output
+        output = _normalize_cli_output(result.output)
+        for value in (
+            self.MARKUP_INTEGRATION["id"],
+            self.MARKUP_INTEGRATION["name"],
+            self.MARKUP_INTEGRATION["version"],
+            self.MARKUP_INTEGRATION["description"],
+            self.MARKUP_INTEGRATION["author"],
+            self.MARKUP_INTEGRATION["license"],
+            self.MARKUP_INTEGRATION["repository"],
+            self.MARKUP_INTEGRATION["tags"][0],
+            self.MARKUP_INTEGRATION["_catalog_name"],
+        ):
+            assert value in output
+
    # -- validation vs network guidance ------------------------------------

    def test_search_local_config_error_shows_local_config_tip(