diff --git a/bcc.py b/bcc.py index c71da94..3c58e89 100644 --- a/bcc.py +++ b/bcc.py @@ -10,6 +10,7 @@ Run: python mcp_manager.py from __future__ import annotations +import html import sys import time from pathlib import Path @@ -26,10 +27,12 @@ from PySide6.QtGui import ( QKeySequence, QPainter, QPixmap, + QTextCursor, ) from PySide6.QtWidgets import ( QAbstractItemView, QApplication, + QButtonGroup, QCheckBox, QComboBox, QDialog, @@ -65,6 +68,26 @@ import bcc_core as core # thread during drag-and-drop import, so skip anything larger than this. MAX_DROP_IMPORT_BYTES = 5 * 1024 * 1024 # 5 MB + +def plain_label(text: object) -> QLabel: + """A QLabel guaranteed to render `text` as plain text, never HTML. + + Qt's QLabel auto-interprets HTML by default (Qt.AutoText). Every catalog + entry field (description, notes, display name, urls -- and especially + args) is attacker-influenceable: catalog.json accepts community PRs, and + only a valid Ed25519 signature stands between a PR and what a user sees + here. A `` or `` in a description must render as + visible text, not markup -- exactly the same reasoning catalog_console.py + documents for its own plain_label(). Every catalog-derived string shown + by the Browse dialog MUST go through this helper (or an inherently + plain-text widget like QPlainTextEdit) rather than a bare QLabel(...). + """ + label = QLabel(html.escape(str(text))) + label.setTextFormat(Qt.TextFormat.PlainText) + label.setWordWrap(True) + return label + + # --- One-line rebrand: change this to recolor the whole app --------------- # ACCENT = "#f97316" # warm orange ACCENT_DIM = "#c2570b" @@ -676,6 +699,39 @@ class ServerEditor(QFrame): def current_name(self) -> str: return self.name.text().strip() + def focus_target(self, target: tuple[str, int | str] | None): + """ + Focus the field a catalog Add left unfilled -- `target` is whatever + core.first_unfilled_focus_target() returned: ("args", line_index), + ("env", var_name), or None (nothing to fill, so do nothing). + + Only meaningful on the stdio page, which is the only page a catalog + entry ever populates (link-only entries never reach dump_data()). + """ + if not target or self.type.currentIndex() != 0: + return + kind, value = target + if kind == "args": + self.args.setFocus() + cursor = self.args.textCursor() + cursor.movePosition(QTextCursor.MoveOperation.Start) + cursor.movePosition( + QTextCursor.MoveOperation.Down, QTextCursor.MoveMode.MoveAnchor, int(value) + ) + cursor.movePosition( + QTextCursor.MoveOperation.EndOfLine, QTextCursor.MoveMode.KeepAnchor + ) + self.args.setTextCursor(cursor) + elif kind == "env": + for r in range(self.env.table.rowCount()): + key_item = self.env.table.item(r, 0) + if key_item and key_item.text() == value: + self.env.table.setCurrentCell(r, 1) + val_item = self.env.table.item(r, 1) + if val_item: + self.env.table.editItem(val_item) + break + def _type_switched(self): self.stack.setCurrentIndex(self.type.currentIndex()) self._emit() @@ -1204,6 +1260,262 @@ class PasteDialog(QDialog): self.err.setText(str(e)) +# --------------------------------------------------------------------------- # +# Browse catalog dialog (issue #10 phase 2): search/filter the signed, +# bundled server catalog and add a "basic" entry through the existing +# paste/import path, or send a "link-only" entry to its setup docs. +# +# Every widget here that shows catalog-derived text uses plain_label() or an +# inherently-plain widget (QPlainTextEdit) -- see plain_label()'s docstring. +# The dialog itself does no signature/schema work: MainWindow hands it an +# already-verified `entries` list (bcc_core.load_bundled_catalog_entries()), +# and an empty list here means "show the empty state", never "fall back to +# something less trusted". +# --------------------------------------------------------------------------- # +class BrowseCatalogDialog(QDialog): + def __init__(self, parent, entries: list[dict]): + super().__init__(parent) + self.setWindowTitle("Browse catalog") + self.resize(880, 560) + self.entries = entries or [] + self.result_entry: dict | None = None + self._current_entry: dict | None = None + self._current_homepage: str | None = None + self._current_docs_url: str | None = None + self._current_group = "All" + + outer = QVBoxLayout(self) + + if not self.entries: + # Signature verification failed, or nothing was bundled -- never + # show a half-trusted list, and never explain WHY beyond this; + # a stale/tampered catalog isn't the user's problem to diagnose. + msg = plain_label("Catalog unavailable.") + msg.setObjectName("placeholder") + msg.setAlignment(Qt.AlignmentFlag.AlignCenter) + outer.addWidget(msg, 1) + btns = QDialogButtonBox(QDialogButtonBox.StandardButton.Close) + btns.rejected.connect(self.reject) + outer.addWidget(btns) + return + + search_row = QHBoxLayout() + self.search_box = QLineEdit() + self.search_box.setPlaceholderText("Search by name, description, or category…") + self.search_box.setClearButtonEnabled(True) + self.search_box.textChanged.connect(self._refresh_list) + search_row.addWidget(self.search_box, 1) + outer.addLayout(search_row) + + chip_row = QHBoxLayout() + self._chip_group = QButtonGroup(self) + self._chip_group.setExclusive(True) + for label in core.CATALOG_CATEGORY_CHIPS: + btn = QPushButton(label) + btn.setCheckable(True) + btn.setChecked(label == "All") + btn.clicked.connect(lambda _checked=False, g=label: self._set_group(g)) + self._chip_group.addButton(btn) + chip_row.addWidget(btn) + chip_row.addStretch() + outer.addLayout(chip_row) + + splitter = QSplitter(Qt.Orientation.Horizontal) + + left = QWidget() + lv = QVBoxLayout(left) + lv.setContentsMargins(0, 0, 0, 0) + self.list = QListWidget() + self.list.currentItemChanged.connect(self._on_selected) + lv.addWidget(self.list, 1) + splitter.addWidget(left) + + right = QFrame() + right.setObjectName("card") + rv = QVBoxLayout(right) + self.detail_title = plain_label("") + self.detail_title.setObjectName("h1") + rv.addWidget(self.detail_title) + self.detail_meta = plain_label("") + self.detail_meta.setObjectName("muted") + rv.addWidget(self.detail_meta) + self.detail_freshness = plain_label("") + self.detail_freshness.setObjectName("muted") + rv.addWidget(self.detail_freshness) + self.detail_desc = plain_label("") + rv.addWidget(self.detail_desc) + self.detail_notes = plain_label("") + self.detail_notes.setObjectName("muted") + rv.addWidget(self.detail_notes) + self.detail_homepage_btn = QPushButton("Open homepage") + self.detail_homepage_btn.clicked.connect(self._open_homepage) + rv.addWidget(self.detail_homepage_btn) + rv.addWidget(plain_label("Exact command this will add:")) + self.detail_command_preview = QPlainTextEdit() + self.detail_command_preview.setObjectName("diag") + self.detail_command_preview.setReadOnly(True) + # Read-only QPlainTextEdit never interprets HTML, regardless of what + # a compromised/careless catalog entry's command/args contain -- this + # is the field that renders "the exact bytes that will be written". + rv.addWidget(self.detail_command_preview, 1) + self.detail_action_btn = QPushButton("") + self.detail_action_btn.setObjectName("primary") + self.detail_action_btn.clicked.connect(self._on_action) + rv.addWidget(self.detail_action_btn) + splitter.addWidget(right) + + splitter.setSizes([360, 480]) + outer.addWidget(splitter, 1) + + btns = QDialogButtonBox(QDialogButtonBox.StandardButton.Close) + btns.rejected.connect(self.reject) + outer.addWidget(btns) + + self._refresh_list() + + # --- list / filtering -------------------------------------------------- # + def _set_group(self, group: str): + self._current_group = group + self._refresh_list() + + def _visible_entries(self) -> list[dict]: + filtered = core.filter_catalog_entries(self.entries, self.search_box.text()) + return core.catalog_entries_in_group(filtered, self._current_group) + + def _format_row(self, entry: dict) -> str: + display = str(entry.get("display") or entry.get("id") or "") + official = "✓ " if entry.get("official") else "" + stars = entry.get("stars") + star_txt = f" ★ {stars:,}" if isinstance(stars, int) else "" + group = core.catalog_category_group(entry.get("category", "")) + desc = str(entry.get("description") or "") + if len(desc) > 88: + desc = desc[:87] + "…" + # QListWidgetItem text is always rendered literally by Qt (no HTML + # interpretation), so no escaping is needed here -- unlike QLabel. + return f"{official}{display}{star_txt}\n{desc} · {group}" + + def _refresh_list(self): + self.list.blockSignals(True) + self.list.clear() + for entry in self._visible_entries(): + item = QListWidgetItem(self._format_row(entry)) + item.setData(Qt.ItemDataRole.UserRole, entry) + # Tooltips DO auto-detect rich text in Qt, so escape defensively + # even though descriptions are already shown, unescaped-but-safe, + # in the QListWidgetItem text above. + item.setToolTip(html.escape(str(entry.get("description", "")))) + self.list.addItem(item) + self.list.blockSignals(False) + if self.list.count(): + self.list.setCurrentRow(0) + else: + self._on_selected(None, None) + + # --- detail pane -------------------------------------------------------- # + def _on_selected(self, current, _previous=None): + if current is None: + self._current_entry = None + self._current_homepage = None + self._current_docs_url = None + self.detail_title.setText("") + self.detail_meta.setText("") + self.detail_freshness.setText("") + self.detail_desc.setText("No matching servers." if self.entries else "") + self.detail_notes.setText("") + self.detail_homepage_btn.setVisible(False) + self.detail_command_preview.setPlainText("") + self.detail_action_btn.setEnabled(False) + self.detail_action_btn.setText("Add") + return + + entry = current.data(Qt.ItemDataRole.UserRole) + self._current_entry = entry + self.detail_title.setText(str(entry.get("display") or entry.get("id") or "")) + + official = "✓ Official" if entry.get("official") else "" + stars = entry.get("stars") + star_txt = f"★ {stars:,}" if isinstance(stars, int) else "" + group = core.catalog_category_group(entry.get("category", "")) + meta_bits = [b for b in (official, star_txt, group) if b] + self.detail_meta.setText(" · ".join(meta_bits)) + + freshness = core.format_freshness_hint(entry.get("last_release")) + self.detail_freshness.setText(freshness) + self.detail_freshness.setVisible(bool(freshness)) + + self.detail_desc.setText(str(entry.get("description") or "")) + + notes = entry.get("notes") or "" + self.detail_notes.setText(notes) + self.detail_notes.setVisible(bool(notes)) + + homepage = entry.get("homepage") + self._current_homepage = homepage if isinstance(homepage, str) else None + self.detail_homepage_btn.setVisible(bool(self._current_homepage)) + + self._current_docs_url = ( + entry.get("docs_url") if isinstance(entry.get("docs_url"), str) else None + ) + + self.detail_command_preview.setPlainText(self._render_command_preview(entry)) + + if entry.get("setup") == "basic": + self.detail_action_btn.setText("Add") + self.detail_action_btn.setEnabled(True) + else: + self.detail_action_btn.setText("Open setup docs") + self.detail_action_btn.setEnabled(bool(self._current_docs_url)) + + def _render_command_preview(self, entry: dict) -> str: + """ + The exact command that will be written, rendered verbatim. Every + value here comes straight from the (signature-verified) catalog + entry with no interpretation beyond str() -- this must never be the + place a markup-laced description sneaks back in as "helpful" + formatting. + """ + if entry.get("setup") != "basic": + docs = entry.get("docs_url") or "(none provided)" + return ( + "This is a hosted/managed integration -- there is no local " + "command to add.\n\nSetup docs:\n " + str(docs) + ) + + config = entry.get("config") or {} + lines = [f"command: {config.get('command', '')}"] + args = config.get("args") or [] + if args: + lines.append("args:") + lines.extend(f" {a}" for a in args) + env = config.get("env") or {} + env_required = entry.get("env_required") or {} + env_keys = list(env.keys()) + [k for k in env_required if k not in env] + if env_keys: + lines.append("env (names only -- you provide the values):") + lines.extend(f" {k}" for k in env_keys) + return "\n".join(lines) + + def _open_homepage(self): + url = self._current_homepage + if url and url.startswith("https://"): + QDesktopServices.openUrl(QUrl(url)) + + def _on_action(self): + entry = self._current_entry + if not entry: + return + if entry.get("setup") == "basic": + self.result_entry = entry + self.accept() + else: + url = self._current_docs_url + if url and url.startswith("https://"): + QDesktopServices.openUrl(QUrl(url)) + # link-only never auto-adds and never closes the dialog -- the + # user can keep browsing after opening the docs in their browser. + + # --------------------------------------------------------------------------- # # Log viewer dialog (issue #6): a read-only, auto-tailing view of a single # server's MCP log file. Polls on a QTimer instead of watching the filesystem @@ -1780,6 +2092,10 @@ class MainWindow(QMainWindow): self.del_btn = QPushButton("Delete") self.del_btn.setObjectName("danger") self.paste_btn = QPushButton("Paste JSON...") + self.browse_catalog_btn = QPushButton("Browse catalog…") + self.browse_catalog_btn.setToolTip( + "Add a popular MCP server from the curated, signed catalog" + ) self.copy_btn = QPushButton("Copy to ▸") self.undo_btn = QPushButton("Undo") self.undo_btn.setEnabled(False) @@ -1792,6 +2108,7 @@ class MainWindow(QMainWindow): self.dup_btn.clicked.connect(self.duplicate_server) self.del_btn.clicked.connect(self.delete_server) self.paste_btn.clicked.connect(self.paste_json) + self.browse_catalog_btn.clicked.connect(self.browse_catalog) self.copy_btn.clicked.connect(self.copy_to_menu) self.undo_btn.clicked.connect(self._undo) self.test_all_btn.clicked.connect(self._test_all_servers) @@ -1800,6 +2117,7 @@ class MainWindow(QMainWindow): self.dup_btn, self.del_btn, self.paste_btn, + self.browse_catalog_btn, self.copy_btn, self.undo_btn, self.test_all_btn, @@ -2378,6 +2696,41 @@ class MainWindow(QMainWindow): self._mark_dirty() self.status.setText(f"Imported {added} added, {replaced} replaced. Review and Save.") + def browse_catalog(self): + """ + Open the Browse-catalog dialog (issue #10 phase 2). The catalog is + loaded and signature-verified fresh every time the dialog opens -- + never cached across app runs at this phase (remote fetch/cache is + #61, not yet built) -- so a bundled-catalog swap only takes effect + on next dialog open, never mid-session in a stale way. + """ + entries = core.load_bundled_catalog_entries( + _asset_dir() / "data" / "catalog.json", _asset_dir() / "data" / "catalog.json.sig" + ) + dlg = BrowseCatalogDialog(self, entries) + if dlg.exec() != QDialog.DialogCode.Accepted or not dlg.result_entry: + return + entry = dlg.result_entry + paste = core.catalog_entry_to_paste_json(entry) + name, data = next(iter(paste.items())) + + existing_before = {s.name: i for i, s in enumerate(self.servers)} + self._push_undo() + _added, replaced = self._import_server(name, data) + idx = ( + existing_before.get(name, len(self.servers) - 1) if replaced else len(self.servers) - 1 + ) + self._refresh_tables(select_index=idx) + self._mark_dirty() + + target = core.first_unfilled_focus_target(data) + self.editor.focus_target(target) + + verb = "Replaced" if replaced else "Added" + self.status.setText( + f"{verb} “{name}” from the catalog. Fill in the highlighted field and Save." + ) + def copy_to_menu(self): idx = self._current_index() if not (0 <= idx < len(self.servers)): @@ -2456,6 +2809,29 @@ class MainWindow(QMainWindow): QMessageBox.warning(self, "Can't save yet", "Fix the highlighted problem first.") return + # Placeholder guard (issue #10): a catalog Add can leave a + # -style token in args/env until the user fills it in. + # This warns, it does not block -- the user may be deliberately + # saving a stub to finish later -- but it must never save silently, + # since a server launched with a literal "" argument + # just fails in a confusing way at spawn time. + placeholder_names = [ + s.name for s in self.servers if core.config_has_unfilled_placeholders(s.data) + ] + if placeholder_names: + names = ", ".join(f"“{n}”" for n in placeholder_names) + ans = QMessageBox.warning( + self, + "Unfilled placeholder", + f"{names} still has a value that hasn't been " + "replaced with a real value. Claude won't be able to use " + "it as-is.\n\nSave anyway?", + QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No, + QMessageBox.StandardButton.No, + ) + if ans != QMessageBox.StandardButton.Yes: + return + # Stale-file check: if the file changed on disk since we loaded it, prompt. # Compare mtime AND size (not mtime alone) so a concurrent external write # that lands within the mtime resolution window, or that restores the diff --git a/bcc.spec b/bcc.spec index dfd12b8..58d6de8 100644 --- a/bcc.spec +++ b/bcc.spec @@ -31,7 +31,11 @@ a = Analysis( ["bcc.py"], pathex=[], binaries=[], - datas=[("icons", "icons"), ("data/catalog.json", "data")], + datas=[ + ("icons", "icons"), + ("data/catalog.json", "data"), + ("data/catalog.json.sig", "data"), + ], hiddenimports=[], hookspath=[], hooksconfig={}, diff --git a/bcc_core.py b/bcc_core.py index c81d5ea..38cea9c 100644 --- a/bcc_core.py +++ b/bcc_core.py @@ -29,6 +29,7 @@ import tempfile import threading import time from dataclasses import dataclass +from datetime import date from pathlib import Path from typing import NamedTuple from urllib.parse import urlparse @@ -2768,8 +2769,17 @@ def catalog_entry_to_paste_json(entry: dict) -> dict: """ Convert a basic-tier catalog entry into the {name: {command, args, env}} shape parse_pasted_json()/_import_server() already understand, so the - (future) catalog picker dialog can feed a selection straight into the - existing paste-import path instead of growing a parallel one. + Browse-catalog dialog can feed a selection straight into the existing + paste-import path instead of growing a parallel one. + + `env` is seeded from two sources: config.env (rare -- e.g. grafana's + non-secret GRAFANA_URL) and, for every key in `env_required` not already + present, an empty-string placeholder. env_required is where the seed + data actually keeps its secret VAR NAMES (validate_catalog requires its + values to be "" -- never a real secret); config.env alone, without this, + would silently drop those names on Add for the 9 of 19 seed entries that + need a secret and only declare it via env_required -- the user would + see a server added with no field prompting them for the key it needs. """ config = entry.get("config") or {} name = entry.get("id") or entry.get("display") or "server" @@ -2777,9 +2787,11 @@ def catalog_entry_to_paste_json(entry: dict) -> dict: "command": config.get("command", ""), "args": list(config.get("args") or []), } - env = config.get("env") + env = dict(config.get("env") or {}) + for key in entry.get("env_required") or {}: + env.setdefault(key, "") if env: - data["env"] = dict(env) + data["env"] = env return {str(name): data} @@ -2799,3 +2811,190 @@ def config_has_unfilled_placeholders(cfg: dict) -> bool: if isinstance(env, dict): values.extend(v for v in env.values() if isinstance(v, str)) return any(_PLACEHOLDER_RE.search(v) for v in values) + + +# --------------------------------------------------------------------------- # +# Catalog: Browse-dialog helpers (issue #10 phase 2) +# +# Everything below is pure and GUI-free on purpose (per the design comment on +# #10): the dialog itself should be a thin shell that calls into this module, +# the same relationship bcc.py already has with the rest of bcc_core.py. +# --------------------------------------------------------------------------- # + +# Collapses the 20-value category taxonomy from the catalog research pass +# down to the 7 chips shown in the Browse dialog. Any category not listed +# here (including one a future catalog entry introduces that we don't yet +# know about) falls back to "Other" rather than raising -- an unrecognized +# category must never make an entry disappear from the dialog. +CATALOG_CATEGORY_GROUPS: dict[str, str] = { + "files": "Files & Dev", + "dev": "Files & Dev", + "code-hosting": "Files & Dev", + "browser": "Files & Dev", + "database": "Data", + "data": "Data", + "search": "Search & AI", + "ai": "Search & AI", + "cloud": "Cloud & Infra", + "infra": "Cloud & Infra", + "observability": "Cloud & Infra", + "productivity": "Work", + "communication": "Work", + "crm": "Work", + "finance": "Work", + "design": "Work", + "media": "Home & Personal", + "smart-home": "Home & Personal", + "personal": "Home & Personal", +} + +# Ordered for chip display: "All" first, the 6 named groups next in the order +# given in the #10 design comment, "Other" last as the catch-all. +CATALOG_CATEGORY_CHIPS: tuple[str, ...] = ( + "All", + "Files & Dev", + "Data", + "Search & AI", + "Cloud & Infra", + "Work", + "Home & Personal", + "Other", +) + + +def catalog_category_group(category: str) -> str: + """Collapse a raw catalog `category` value to one of the 7 UI chips. + + Unknown/missing categories map to "Other" -- never raises, never drops + an entry from the list just because its category tag doesn't match one + of the ones known at the time this mapping was written. + """ + return CATALOG_CATEGORY_GROUPS.get(str(category or "").strip().lower(), "Other") + + +def catalog_entry_matches_query(entry: dict, query: str) -> bool: + """ + Case-insensitive substring match against a catalog entry's id, display + name, description, and category. An empty/whitespace-only query matches + everything, so the search box doubles as "no filter" when cleared -- + the same convention server_matches_filter() uses for the main table. + """ + q = (query or "").strip().lower() + if not q: + return True + haystacks = ( + str(entry.get("id", "")), + str(entry.get("display", "")), + str(entry.get("description", "")), + str(entry.get("category", "")), + ) + return any(q in h.lower() for h in haystacks) + + +def filter_catalog_entries(entries: list[dict], query: str) -> list[dict]: + """Return only the catalog entries that match `query` (see + catalog_entry_matches_query).""" + return [e for e in entries if catalog_entry_matches_query(e, query)] + + +def catalog_entries_in_group(entries: list[dict], group: str) -> list[dict]: + """ + Return only the entries whose category collapses into `group` (one of + CATALOG_CATEGORY_CHIPS). "All" (or a falsy/unrecognized group) returns + every entry unfiltered -- that's the default chip state. + """ + if not group or group == "All": + return list(entries) + return [e for e in entries if catalog_category_group(e.get("category", "")) == group] + + +def format_freshness_hint(last_release: str | None, today: date | None = None) -> str: + """ + Turn a catalog entry's `last_release` (an ISO "YYYY-MM-DD" date, or None + when the registry didn't expose one) into a short freshness hint for the + detail pane, e.g. "Last updated 14 months ago". + + Returns "" (nothing to show) when `last_release` is missing or + unparseable, or when it's somehow in the future relative to `today` -- + a bogus "-3 months ago" would undermine the one signal this hint exists + to give the user, so we'd rather show nothing than something wrong. + + `today` is an injectable override so this is exactly reproducible in + tests without depending on the wall clock. + """ + if not last_release or not isinstance(last_release, str): + return "" + try: + released = date.fromisoformat(last_release) + except ValueError: + return "" + now = today or date.today() + if released > now: + return "" + + months = (now.year - released.year) * 12 + (now.month - released.month) + if now.day < released.day: + months -= 1 + months = max(months, 0) + + if months == 0: + return "Last updated this month" + if months == 1: + return "Last updated 1 month ago" + if months < 24: + return f"Last updated {months} months ago" + years = months // 12 + return f"Last updated {years} year{'s' if years != 1 else ''} ago" + + +def first_unfilled_focus_target(data: dict) -> tuple[str, int | str] | None: + """ + Given a server config dict shaped like catalog_entry_to_paste_json()'s + output (command/args/env), find the first thing a user must fill in + after a catalog Add: a -style arg (checked first, since a + missing path/target usually blocks the server from starting at all) or + else the first env var the catalog left blank. + + Returns ("args", index) or ("env", key), or None when there's nothing + left to fill (e.g. a server with no placeholders and no required env). + The GUI uses this to focus+select the right field right after Add, + instead of leaving the user to hunt for what still needs a value. + """ + args = data.get("args") or [] + for i, a in enumerate(args): + if isinstance(a, str) and _PLACEHOLDER_RE.search(a): + return ("args", i) + + env = data.get("env") or {} + if isinstance(env, dict): + for k, v in env.items(): + if not isinstance(v, str) or not v.strip() or _PLACEHOLDER_RE.search(v): + return ("env", k) + + return None + + +def load_bundled_catalog_entries(catalog_path: Path, sig_path: Path) -> list[dict]: + """ + Read+verify+validate the bundled catalog.json/.sig pair from disk and + return its `servers` list -- or an EMPTY list if anything at all is + wrong: files missing/unreadable, signature doesn't verify, JSON doesn't + parse, or validate_catalog() finds a problem. + + 🔴 SECURITY: this is the load-bearing guarantee for the Browse dialog. + There is deliberately no partial-success path here -- a signature + failure must never surface a half-trusted list, only an empty one, so + the GUI's only job is to render "Catalog unavailable" when this comes + back empty. All the real trust decisions (signature, schema, command + allowlist) already live in resolve_catalog()/validate_catalog(); this + is a thin disk-reading wrapper around them so the GUI never touches + catalog bytes directly. + """ + try: + raw = catalog_path.read_bytes() + sig = sig_path.read_bytes() + except OSError: + return [] + data = resolve_catalog(bundled=(raw, sig), cached=None, remote=None) + servers = data.get("servers") if isinstance(data, dict) else None + return servers if isinstance(servers, list) else [] diff --git a/tests/test_core.py b/tests/test_core.py index 725358f..98562a9 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -2357,6 +2357,43 @@ def test_catalog_entry_to_paste_json_includes_env_when_present(): assert c.validate_catalog(malicious) != [] +def test_catalog_entry_to_paste_json_seeds_env_required_keys(): + # Regression: env_required is where the seed data actually keeps its + # secret VAR NAMES (postgres/github/notion/etc. all declare their secret + # here with config.env left empty) -- catalog_entry_to_paste_json must + # surface those names as blank env rows, not silently drop them. + entry = _minimal_catalog()["servers"][0] + entry["env_required"] = {"DATABASE_URI": ""} + result = c.catalog_entry_to_paste_json(entry) + assert result["widget"]["env"] == {"DATABASE_URI": ""} + + +def test_catalog_entry_to_paste_json_config_env_wins_over_env_required_default(): + entry = _minimal_catalog()["servers"][0] + entry["config"]["env"] = {"GRAFANA_URL": ""} + entry["env_required"] = {"GRAFANA_URL": "", "GRAFANA_SERVICE_ACCOUNT_TOKEN": ""} + result = c.catalog_entry_to_paste_json(entry) + assert result["widget"]["env"] == { + "GRAFANA_URL": "", + "GRAFANA_SERVICE_ACCOUNT_TOKEN": "", + } + + +def test_catalog_entry_to_paste_json_real_postgres_entry_seeds_database_uri(): + """End-to-end regression against the actual shipped postgres entry, + which needs DATABASE_URI via env_required and has no config.env at + all -- this is exactly the shape that was silently dropping the env + field before catalog_entry_to_paste_json accounted for env_required.""" + root = Path(__file__).resolve().parent.parent + raw = (root / "data" / "catalog.json").read_bytes() + data = c.load_catalog(raw) + entry = next(s for s in data["servers"] if s["id"] == "postgres") + result = c.catalog_entry_to_paste_json(entry) + assert result["postgres"]["env"] == {"DATABASE_URI": ""} + # And the focus-target helper now has something to point the user at. + assert c.first_unfilled_focus_target(result["postgres"]) == ("env", "DATABASE_URI") + + def test_config_has_unfilled_placeholders_true_for_token(): cfg = {"command": "npx", "args": ["-y", "server", ""]} assert c.config_has_unfilled_placeholders(cfg) is True @@ -2370,3 +2407,310 @@ def test_config_has_unfilled_placeholders_false_after_fill(): def test_config_has_unfilled_placeholders_checks_env_too(): cfg = {"command": "uvx", "args": ["mcp-grafana"], "env": {"GRAFANA_URL": ""}} assert c.config_has_unfilled_placeholders(cfg) is True + + +# --------------------------------------------------------------------------- # +# Browse-catalog dialog helpers (issue #10 phase 2) +# --------------------------------------------------------------------------- # + + +# --- catalog_category_group / CATALOG_CATEGORY_GROUPS --------------------- # +@pytest.mark.parametrize( + "category,expected_group", + [ + ("files", "Files & Dev"), + ("dev", "Files & Dev"), + ("code-hosting", "Files & Dev"), + ("browser", "Files & Dev"), + ("database", "Data"), + ("data", "Data"), + ("search", "Search & AI"), + ("ai", "Search & AI"), + ("cloud", "Cloud & Infra"), + ("infra", "Cloud & Infra"), + ("observability", "Cloud & Infra"), + ("productivity", "Work"), + ("communication", "Work"), + ("crm", "Work"), + ("finance", "Work"), + ("design", "Work"), + ("media", "Home & Personal"), + ("smart-home", "Home & Personal"), + ("personal", "Home & Personal"), + ], +) +def test_catalog_category_group_maps_every_taxonomy_value(category, expected_group): + assert c.catalog_category_group(category) == expected_group + + +def test_catalog_category_group_unknown_falls_back_to_other(): + assert c.catalog_category_group("some-future-category-nobody-has-seen-yet") == "Other" + assert c.catalog_category_group("") == "Other" + assert c.catalog_category_group(None) == "Other" + + +def test_catalog_category_group_is_case_insensitive(): + assert c.catalog_category_group("Files") == "Files & Dev" + assert c.catalog_category_group("DATABASE") == "Data" + + +def test_shipped_catalog_categories_all_have_a_known_group(): + """Regression: every category actually used in data/catalog.json must + collapse to one of the 7 chips, never silently drop an entry.""" + root = Path(__file__).resolve().parent.parent + raw = (root / "data" / "catalog.json").read_bytes() + data = c.load_catalog(raw) + for entry in data["servers"]: + group = c.catalog_category_group(entry["category"]) + assert group in c.CATALOG_CATEGORY_CHIPS + + +# --- catalog_entry_matches_query / filter_catalog_entries ------------------ # +def _catalog_entries(): + return [ + { + "id": "filesystem", + "display": "Filesystem", + "description": "Read/write access to local directories you choose.", + "category": "files", + }, + { + "id": "postgres", + "display": "Postgres MCP Pro", + "description": "Query and inspect a PostgreSQL database.", + "category": "database", + }, + { + "id": "slack", + "display": "Slack", + "description": "Search messages and send messages from your assistant.", + "category": "communication", + }, + ] + + +def test_catalog_entry_matches_query_empty_matches_everything(): + entries = _catalog_entries() + assert c.filter_catalog_entries(entries, "") == entries + assert c.filter_catalog_entries(entries, " ") == entries + + +def test_catalog_entry_matches_query_matches_id(): + result = c.filter_catalog_entries(_catalog_entries(), "postgres") + assert [e["id"] for e in result] == ["postgres"] + + +def test_catalog_entry_matches_query_matches_display_case_insensitive(): + result = c.filter_catalog_entries(_catalog_entries(), "SLACK") + assert [e["id"] for e in result] == ["slack"] + + +def test_catalog_entry_matches_query_matches_description(): + result = c.filter_catalog_entries(_catalog_entries(), "PostgreSQL database") + assert [e["id"] for e in result] == ["postgres"] + + +def test_catalog_entry_matches_query_matches_category(): + result = c.filter_catalog_entries(_catalog_entries(), "database") + assert [e["id"] for e in result] == ["postgres"] + + +def test_catalog_entry_matches_query_no_match_returns_empty(): + assert c.filter_catalog_entries(_catalog_entries(), "kubernetes") == [] + + +def test_catalog_entries_in_group_all_returns_everything(): + entries = _catalog_entries() + assert c.catalog_entries_in_group(entries, "All") == entries + assert c.catalog_entries_in_group(entries, "") == entries + assert c.catalog_entries_in_group(entries, None) == entries + + +def test_catalog_entries_in_group_filters_by_collapsed_category(): + result = c.catalog_entries_in_group(_catalog_entries(), "Data") + assert [e["id"] for e in result] == ["postgres"] + + result = c.catalog_entries_in_group(_catalog_entries(), "Work") + assert [e["id"] for e in result] == ["slack"] + + +# --- format_freshness_hint -------------------------------------------------- # +def test_format_freshness_hint_none_returns_empty_string(): + assert c.format_freshness_hint(None) == "" + assert c.format_freshness_hint("") == "" + + +def test_format_freshness_hint_unparseable_returns_empty_string(): + assert c.format_freshness_hint("not-a-date") == "" + + +def test_format_freshness_hint_this_month(): + assert ( + c.format_freshness_hint("2026-07-01", today=c.date(2026, 7, 12)) + == "Last updated this month" + ) + + +def test_format_freshness_hint_one_month_singular(): + assert ( + c.format_freshness_hint("2026-06-01", today=c.date(2026, 7, 12)) + == "Last updated 1 month ago" + ) + + +def test_format_freshness_hint_months_ago(): + # Exactly 14 full months elapsed, no day-of-month remainder to round off. + assert ( + c.format_freshness_hint("2025-01-15", today=c.date(2026, 3, 15)) + == "Last updated 14 months ago" + ) + + +def test_format_freshness_hint_rounds_down_partial_month(): + # 2025-05-16 -> 2026-07-12 is 13 full months, not 14: the 14th month + # would only complete on 2026-07-16. + assert ( + c.format_freshness_hint("2025-05-16", today=c.date(2026, 7, 12)) + == "Last updated 13 months ago" + ) + + +def test_format_freshness_hint_years_ago(): + assert ( + c.format_freshness_hint("2024-01-01", today=c.date(2026, 7, 12)) + == "Last updated 2 years ago" + ) + + +def test_format_freshness_hint_23_months_stays_in_months_not_years(): + # The switch to "N years ago" happens at 24 full months, not 12 -- the + # whole point of this hint is the granular "14 months ago" phrasing the + # design comment on #10 asked for, so 13-23 months must stay in months. + assert ( + c.format_freshness_hint("2024-08-12", today=c.date(2026, 7, 12)) + == "Last updated 23 months ago" + ) + + +def test_format_freshness_hint_future_date_returns_empty_string(): + # A last_release "in the future" relative to `today` is nonsensical -- + # show nothing rather than a misleading negative offset. + assert c.format_freshness_hint("2027-01-01", today=c.date(2026, 7, 12)) == "" + + +# --- first_unfilled_focus_target -------------------------------------------- # +def test_first_unfilled_focus_target_prefers_placeholder_arg(): + data = { + "command": "npx", + "args": ["-y", "server", ""], + "env": {"API_KEY": ""}, + } + assert c.first_unfilled_focus_target(data) == ("args", 2) + + +def test_first_unfilled_focus_target_falls_back_to_first_blank_env(): + data = {"command": "uvx", "args": ["mcp-grafana"], "env": {"GRAFANA_URL": ""}} + assert c.first_unfilled_focus_target(data) == ("env", "GRAFANA_URL") + + +def test_first_unfilled_focus_target_none_when_fully_filled(): + data = {"command": "npx", "args": ["-y", "server"], "env": {"API_KEY": "sk-real-value"}} + assert c.first_unfilled_focus_target(data) is None + + +def test_first_unfilled_focus_target_none_for_config_with_no_env_or_args(): + assert c.first_unfilled_focus_target({"command": "npx", "args": []}) is None + + +# --- load_bundled_catalog_entries ------------------------------------------- # +def test_load_bundled_catalog_entries_valid_signature(tmp_path, monkeypatch): + priv = Ed25519PrivateKey.generate() + pub = priv.public_key().public_bytes_raw() + monkeypatch.setattr(c, "CATALOG_PUBKEYS", [pub]) + + raw, sig = _signed(_minimal_catalog(version=1), priv) + catalog_path = tmp_path / "catalog.json" + sig_path = tmp_path / "catalog.json.sig" + catalog_path.write_bytes(raw) + sig_path.write_bytes(sig) + + entries = c.load_bundled_catalog_entries(catalog_path, sig_path) + assert len(entries) == 1 + assert entries[0]["id"] == "widget" + + +def test_load_bundled_catalog_entries_tampered_payload_returns_empty_list(tmp_path, monkeypatch): + priv = Ed25519PrivateKey.generate() + pub = priv.public_key().public_bytes_raw() + monkeypatch.setattr(c, "CATALOG_PUBKEYS", [pub]) + + raw, sig = _signed(_minimal_catalog(version=1), priv) + tampered = bytearray(raw) + tampered[-2] ^= 0xFF # flip a byte inside the trailing bytes, still valid-ish JSON shape + catalog_path = tmp_path / "catalog.json" + sig_path = tmp_path / "catalog.json.sig" + catalog_path.write_bytes(bytes(tampered)) + sig_path.write_bytes(sig) + + assert c.load_bundled_catalog_entries(catalog_path, sig_path) == [] + + +def test_load_bundled_catalog_entries_wrong_key_returns_empty_list(tmp_path, monkeypatch): + priv = Ed25519PrivateKey.generate() + other_priv = Ed25519PrivateKey.generate() + other_pub = other_priv.public_key().public_bytes_raw() + monkeypatch.setattr(c, "CATALOG_PUBKEYS", [other_pub]) + + raw, sig = _signed(_minimal_catalog(version=1), priv) # signed by the WRONG key + catalog_path = tmp_path / "catalog.json" + sig_path = tmp_path / "catalog.json.sig" + catalog_path.write_bytes(raw) + sig_path.write_bytes(sig) + + assert c.load_bundled_catalog_entries(catalog_path, sig_path) == [] + + +def test_load_bundled_catalog_entries_missing_files_returns_empty_list(tmp_path): + assert c.load_bundled_catalog_entries(tmp_path / "nope.json", tmp_path / "nope.json.sig") == [] + + +def test_load_bundled_catalog_entries_missing_sig_returns_empty_list(tmp_path, monkeypatch): + priv = Ed25519PrivateKey.generate() + pub = priv.public_key().public_bytes_raw() + monkeypatch.setattr(c, "CATALOG_PUBKEYS", [pub]) + + raw, _sig = _signed(_minimal_catalog(version=1), priv) + catalog_path = tmp_path / "catalog.json" + catalog_path.write_bytes(raw) + missing_sig_path = tmp_path / "catalog.json.sig" # never written + + assert c.load_bundled_catalog_entries(catalog_path, missing_sig_path) == [] + + +def test_load_bundled_catalog_entries_invalid_but_signed_returns_empty_list(tmp_path, monkeypatch): + """A payload that verifies but fails validate_catalog() (disallowed + command) must still come back empty -- signing is necessary, not + sufficient.""" + priv = Ed25519PrivateKey.generate() + pub = priv.public_key().public_bytes_raw() + monkeypatch.setattr(c, "CATALOG_PUBKEYS", [pub]) + + raw, sig = _signed(_catalog_with({"config": {"command": "bash", "args": []}}), priv) + catalog_path = tmp_path / "catalog.json" + sig_path = tmp_path / "catalog.json.sig" + catalog_path.write_bytes(raw) + sig_path.write_bytes(sig) + + assert c.load_bundled_catalog_entries(catalog_path, sig_path) == [] + + +def test_load_bundled_catalog_entries_real_shipped_catalog(): + """End-to-end regression against the actual bundled data/catalog.json + + .sig, using the real CATALOG_PUBKEYS (no monkeypatch) -- this is what + the Browse dialog actually calls on startup.""" + root = Path(__file__).resolve().parent.parent + entries = c.load_bundled_catalog_entries( + root / "data" / "catalog.json", root / "data" / "catalog.json.sig" + ) + assert len(entries) == 19 + assert {e["id"] for e in entries} >= {"filesystem", "github", "slack", "postgres"}