6 Commits

Author SHA1 Message Date
the_og cfe05b4324 Merge branch 'main' into feat/10-browse-dialog
CI / Tests (py3.12 / windows-latest) (pull_request) Failing after 23s
CI / Lint (ruff) (pull_request) Successful in 6s
CI / Tests (py3.10 / ubuntu-latest) (pull_request) Successful in 10s
CI / Tests (py3.12 / ubuntu-latest) (pull_request) Successful in 10s
CI / Tests (py3.13 / ubuntu-latest) (pull_request) Successful in 10s
CI / Catalog signature (pull_request) Successful in 7s
2026-07-12 21:30:19 -04:00
the_og cd2ac2f6f8 Merge PR #69: make the review gate load-bearing, split the keys (#68)
CI / Tests (py3.10 / ubuntu-latest) (push) Successful in 10s
CI / Tests (py3.12 / windows-latest) (push) Successful in 24s
CI / Lint (ruff) (push) Successful in 6s
CI / Tests (py3.12 / ubuntu-latest) (push) Successful in 11s
CI / Tests (py3.13 / ubuntu-latest) (push) Successful in 11s
CI / Catalog signature (push) Successful in 7s
Finding 1: can_sign() returned True on an empty changeset, and _on_sign
compared the reviewed blob against a hardcoded "main" rather than the ref
actually reviewed — so the PR path could never sign, and the main-vs-main
path unlocked Sign with zero entries acknowledged. That is how commit b08cf21
signed 19 entries nobody reviewed. can_sign now refuses an empty diff, checks
has_blocking_risk itself instead of trusting a GUI checkbox, and resolves the
TOCTOU pin from the reviewed ref via a pure, testable sign_precondition().

Finding 5: the catalog key and the release key are now separate. The release
key lives in CI and signs checksums; the catalog key stays offline and signs
what users execute. A CI compromise gets the former, not the latter.
2026-07-12 21:30:13 -04:00
the_og 26c66b7db1 Merge branch 'main' into fix/68-console-gate
CI / Tests (py3.12 / windows-latest) (pull_request) Successful in 24s
CI / Lint (ruff) (pull_request) Successful in 6s
CI / Tests (py3.10 / ubuntu-latest) (pull_request) Successful in 10s
CI / Tests (py3.12 / ubuntu-latest) (pull_request) Successful in 10s
CI / Tests (py3.13 / ubuntu-latest) (pull_request) Successful in 10s
CI / Catalog signature (pull_request) Successful in 6s
2026-07-12 21:28:27 -04:00
the_og 86139100eb Merge PR #70: enforce the checks we said we had (#68)
CI / Tests (py3.12 / windows-latest) (push) Successful in 23s
CI / Lint (ruff) (push) Successful in 6s
CI / Tests (py3.10 / ubuntu-latest) (push) Successful in 11s
CI / Tests (py3.12 / ubuntu-latest) (push) Successful in 10s
CI / Tests (py3.13 / ubuntu-latest) (push) Successful in 10s
CI / Catalog signature (push) Successful in 6s
Findings 2/3/4/6/7. config.env now gets the deny-list, ASCII, secret and
empty-or-placeholder checks that args always had; version pinning is enforced
at runtime, not only in the maintainer tool; the CI gate pins the expected
pubkey instead of trusting the one in the PR it is reviewing; resolve_catalog
anchors its cap to the bundled version and prefers bundled on ties; catalog
ids are constrained.

Tests rewritten: the old fixtures asserted the unpinned form validates clean
and that config.env passes through verbatim — they enshrined two of the bugs.
2026-07-12 21:28:24 -04:00
BCC Agent 38f14deeff fix(core): validate config.env, enforce version pinning, fix CI trust anchor and resolve_catalog guards (#68)
CI / Lint (ruff) (pull_request) Successful in 7s
CI / Tests (py3.10 / ubuntu-latest) (pull_request) Successful in 11s
CI / Tests (py3.12 / ubuntu-latest) (pull_request) Successful in 11s
CI / Tests (py3.13 / ubuntu-latest) (pull_request) Successful in 10s
CI / Tests (py3.12 / windows-latest) (pull_request) Successful in 23s
CI / Catalog signature (pull_request) Successful in 7s
Fixes findings 2, 3, 4, 6, 7 from the issue #68 adversarial review.

- Finding 2: config.env was type-checked only. Add CATALOG_DENIED_ENV_KEYS
  (case-insensitive) for interpreter/loader-override keys (NODE_OPTIONS,
  PYTHONPATH, LD_PRELOAD, ...), apply the ASCII check and the existing
  secret-value check to env keys/values, and require env values to be
  empty or a single <PLACEHOLDER> token.

- Finding 3: version pinning was only checked by catalog_review.py (which
  never runs on the signing path per finding 1). Move enforcement into
  _validate_catalog_config: npm/uvx specs must carry @version or ==version
  (scoped names handled), docker images must have an explicit non-latest
  tag. Only the first plausible package-spec token is checked, so flags,
  <PLACEHOLDER>s, and docker subcommands/flags don't trip it. All 19 real
  catalog entries still validate clean.

- Finding 4: the CI catalog-signature gate imported bcc_core from the PR
  branch and trusted whatever CATALOG_PUBKEYS said there, so a PR changing
  both catalog.json and CATALOG_PUBKEYS (with a matching signature) went
  green. ci.yml now hardcodes the expected base64 pubkey and asserts
  bcc_core.CATALOG_PUBKEYS matches it before verifying the signature.
  NOTE: the maintainer is planning to rotate this key -- update
  EXPECTED_CATALOG_PUBKEY_B64 in ci.yml as its own reviewed change when
  that happens, never bundled with a catalog content change.

- Finding 6: resolve_catalog's anti-rollback/anti-freeze guards sat behind
  `if best_version >= 0`, so the first verified candidate was accepted
  unconditionally and the anti-freeze anchor drifted with each accepted
  candidate instead of staying fixed. The cap is now measured against the
  bundled catalog's version specifically (the trust anchor baked into the
  binary), regardless of evaluation order; bundled wins version ties; and
  a new pure `floor` parameter lets a future caller pass a persisted
  accepted-version floor.

- Finding 7: catalog id is now constrained to ^[a-z0-9][a-z0-9._-]{0,63}$.

Tests: fixed _minimal_catalog to use a pinned package (was enshrining
finding 3), rewrote the env-passthrough test to prove the validation
boundary instead of asserting env passes through unchecked, and
reordered test_resolve_catalog_rejects_absurd_version_jump so it
actually exercises the first-candidate path. Added positive/negative
tests for every new rule. Manually verified each new check by commenting
it out and confirming the guarding test goes red, then restoring it.
2026-07-12 21:27:08 -04:00
Cowork Supervisor dffa0e152f Browse catalog dialog (issue #10 phase 2)
CI / Tests (py3.10 / ubuntu-latest) (pull_request) Successful in 11s
CI / Tests (py3.12 / windows-latest) (pull_request) Failing after 23s
CI / Lint (ruff) (pull_request) Successful in 7s
CI / Tests (py3.12 / ubuntu-latest) (pull_request) Successful in 11s
CI / Tests (py3.13 / ubuntu-latest) (pull_request) Successful in 11s
CI / Catalog signature (pull_request) Successful in 7s
Adds the user-facing half of the server catalog: a searchable, signed
catalog browser wired to the existing paste/import path.

bcc_core.py (pure, GUI-free, per the design comment on #10):
- CATALOG_CATEGORY_GROUPS / catalog_category_group(): collapses the
  raw taxonomy to the 7 UI chips (unknown categories fall back to
  Other, never drop an entry).
- catalog_entry_matches_query() / filter_catalog_entries() /
  catalog_entries_in_group(): search-as-you-type + category chip
  filtering over catalog entries.
- format_freshness_hint(): last_release ISO date -> "Last updated N
  months/years ago", '' when missing/unparseable/future.
- first_unfilled_focus_target(): finds the first <PLACEHOLDER> arg or
  blank env_required key so the GUI can focus it after Add.
- load_bundled_catalog_entries(): reads+verifies+validates the
  bundled catalog.json/.sig pair, returns servers or an EMPTY list on
  ANY failure -- the load-bearing guarantee behind the dialog's empty
  state.
- Bug fix: catalog_entry_to_paste_json() only copied config.env,
  ignoring env_required -- the field where 9 of the 19 seed entries
  (postgres, github, notion, obsidian, brave-search, tavily,
  home-assistant, n8n, grafana) actually declare their secret var
  names. Add would have silently added these servers with no env
  field for the user to fill in. Now env_required keys are seeded as
  empty-string placeholders unless config.env already sets them.

bcc.py:
- BrowseCatalogDialog: search box, category chips, list, detail pane
  (description/notes/homepage/freshness hint/verbatim command
  preview), per-tier action (Add for basic, Open setup docs for
  link-only). Every catalog-derived string goes through plain_label()
  (Qt.PlainText + html.escape, matching catalog_console.py's
  approach) or an inherently-plain QPlainTextEdit for the command
  preview -- catalog.json takes community PRs, so every field is
  attacker-influenceable.
- "Browse catalog…" button next to Paste JSON.
- ServerEditor.focus_target(): selects the first unfilled arg line or
  opens the env-table cell editor for the first blank env row.
- Save-time placeholder guard: warns (Yes/No, defaults No) when any
  server still has an unfilled <PLACEHOLDER>; never blocks a
  deliberate save, never saves one unnoticed.

bcc.spec: bundle data/catalog.json.sig alongside catalog.json -- the
signature file was missing from datas, so a frozen build would have
had a catalog.json with no matching .sig for resolve_catalog() to
verify against.

tests/test_core.py: +82 tests covering category-group mapping,
catalog search/filter, freshness-hint formatting (including the
month/year rounding boundary), first_unfilled_focus_target, the
catalog_entry_to_paste_json env_required fix (incl. a regression
against the real shipped postgres entry), and
load_bundled_catalog_entries under valid/tampered/wrong-key/missing-
file/invalid-but-signed conditions plus an end-to-end check against
the real data/catalog.json + .sig (19 entries).

GUI code (BrowseCatalogDialog, ServerEditor.focus_target,
MainWindow.browse_catalog) is unavoidably untested here -- PySide6
cannot import in this sandbox (missing libEGL/system GL libs) -- but
all the logic it depends on is pushed into bcc_core.py and covered
above.
2026-07-12 19:00:54 -04:00
5 changed files with 1589 additions and 51 deletions
+50 -2
View File
@@ -95,12 +95,39 @@ jobs:
- name: Install dependencies - name: Install dependencies
run: pip install cryptography run: pip install cryptography
# 🔴 TRUST ANCHOR — issue #68 finding 4.
#
# This step used to do `import bcc_core as c` FROM THE CHECKED-OUT PR
# BRANCH and verify the catalog against c.CATALOG_PUBKEYS — i.e. it
# trusted the public key shipped in the very diff it was reviewing. A
# PR that changed data/catalog.json AND bcc_core.CATALOG_PUBKEYS (to
# an attacker key, with a matching signature produced by the attacker's
# matching private key) went green, because there was nothing outside
# the PR's own content to check the key against. The gate's whole
# point is catching a friendly-looking PR the maintainer merges
# without really reading it — and that hole made it a two-file diff.
#
# EXPECTED_CATALOG_PUBKEY_B64 below is hardcoded HERE, in the workflow
# file, independent of whatever bcc_core.py says on the PR branch. It
# is intentionally the only line in this step that matters for
# security review: changing it changes what this gate is willing to
# trust. THIS CONSTANT IS A TRUST ANCHOR. A PR that changes this line
# in the same diff as a catalog change is exactly the attack this gate
# exists to prevent — review a change to this line on its own,
# never bundled with a catalog update.
#
# NOTE for the next key rotation: update EXPECTED_CATALOG_PUBKEY_B64
# below to the new key's base64 form, as its own reviewed change.
- name: Verify data/catalog.json.sig - name: Verify data/catalog.json.sig
env:
EXPECTED_CATALOG_PUBKEY_B64: "082NOwVB7uURkvfyS3+knJ+40Fk6C9unsF47+2uPKo4="
run: | run: |
python - <<'PY' python - <<'PY'
import pathlib, sys import base64, os, pathlib, sys
import bcc_core as c import bcc_core as c
expected_pubkey_b64 = os.environ["EXPECTED_CATALOG_PUBKEY_B64"]
raw = pathlib.Path("data/catalog.json").read_bytes() raw = pathlib.Path("data/catalog.json").read_bytes()
sig_path = pathlib.Path("data/catalog.json.sig") sig_path = pathlib.Path("data/catalog.json.sig")
@@ -111,6 +138,26 @@ jobs:
if b"\x00" * 32 in c.CATALOG_PUBKEYS: if b"\x00" * 32 in c.CATALOG_PUBKEYS:
sys.exit("FAIL: CATALOG_PUBKEYS still holds the placeholder key.") sys.exit("FAIL: CATALOG_PUBKEYS still holds the placeholder key.")
# Trust anchor check FIRST, before verifying anything against
# bcc_core.CATALOG_PUBKEYS: a PR is not allowed to bring its own
# key. CATALOG_PUBKEYS on the checked-out branch must be EXACTLY
# the key(s) this workflow file itself expects -- no more, no
# fewer, no substitutions.
actual_pubkeys_b64 = [base64.b64encode(k).decode() for k in c.CATALOG_PUBKEYS]
if actual_pubkeys_b64 != [expected_pubkey_b64]:
sys.exit(
"FAIL: bcc_core.CATALOG_PUBKEYS on this branch does not match the "
"trust anchor hardcoded in .github/workflows/ci.yml.\n"
f" expected: {[expected_pubkey_b64]}\n"
f" actual: {actual_pubkeys_b64}\n"
"\n"
"This PR is changing (or has changed) the catalog signing key. That "
"change must be reviewed on its own, separately from any catalog "
"content change, and the workflow's EXPECTED_CATALOG_PUBKEY_B64 "
"updated deliberately -- not accepted because it happened to match "
"whatever bcc_core.py says on this branch."
)
if not c.verify_catalog_signature(raw, sig_path.read_bytes(), c.CATALOG_PUBKEYS): if not c.verify_catalog_signature(raw, sig_path.read_bytes(), c.CATALOG_PUBKEYS):
sys.exit( sys.exit(
"FAIL: data/catalog.json does NOT match its signature.\n" "FAIL: data/catalog.json does NOT match its signature.\n"
@@ -124,5 +171,6 @@ jobs:
if problems: if problems:
sys.exit("FAIL: catalog failed validation:\n " + "\n ".join(problems)) sys.exit("FAIL: catalog failed validation:\n " + "\n ".join(problems))
print("OK: catalog signature verifies and the catalog validates clean.") print("OK: catalog signature verifies, the pubkey matches the CI trust anchor, "
"and the catalog validates clean.")
PY PY
+376
View File
@@ -10,6 +10,7 @@ Run: python mcp_manager.py
from __future__ import annotations from __future__ import annotations
import html
import sys import sys
import time import time
from pathlib import Path from pathlib import Path
@@ -26,10 +27,12 @@ from PySide6.QtGui import (
QKeySequence, QKeySequence,
QPainter, QPainter,
QPixmap, QPixmap,
QTextCursor,
) )
from PySide6.QtWidgets import ( from PySide6.QtWidgets import (
QAbstractItemView, QAbstractItemView,
QApplication, QApplication,
QButtonGroup,
QCheckBox, QCheckBox,
QComboBox, QComboBox,
QDialog, QDialog,
@@ -65,6 +68,26 @@ import bcc_core as core
# thread during drag-and-drop import, so skip anything larger than this. # thread during drag-and-drop import, so skip anything larger than this.
MAX_DROP_IMPORT_BYTES = 5 * 1024 * 1024 # 5 MB 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 `<b>` or `<img onerror=...>` 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 --------------- # # --- One-line rebrand: change this to recolor the whole app --------------- #
ACCENT = "#f97316" # warm orange ACCENT = "#f97316" # warm orange
ACCENT_DIM = "#c2570b" ACCENT_DIM = "#c2570b"
@@ -676,6 +699,39 @@ class ServerEditor(QFrame):
def current_name(self) -> str: def current_name(self) -> str:
return self.name.text().strip() 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): def _type_switched(self):
self.stack.setCurrentIndex(self.type.currentIndex()) self.stack.setCurrentIndex(self.type.currentIndex())
self._emit() self._emit()
@@ -1204,6 +1260,262 @@ class PasteDialog(QDialog):
self.err.setText(str(e)) 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 # 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 # 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 = QPushButton("Delete")
self.del_btn.setObjectName("danger") self.del_btn.setObjectName("danger")
self.paste_btn = QPushButton("Paste JSON...") 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.copy_btn = QPushButton("Copy to ▸")
self.undo_btn = QPushButton("Undo") self.undo_btn = QPushButton("Undo")
self.undo_btn.setEnabled(False) self.undo_btn.setEnabled(False)
@@ -1792,6 +2108,7 @@ class MainWindow(QMainWindow):
self.dup_btn.clicked.connect(self.duplicate_server) self.dup_btn.clicked.connect(self.duplicate_server)
self.del_btn.clicked.connect(self.delete_server) self.del_btn.clicked.connect(self.delete_server)
self.paste_btn.clicked.connect(self.paste_json) 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.copy_btn.clicked.connect(self.copy_to_menu)
self.undo_btn.clicked.connect(self._undo) self.undo_btn.clicked.connect(self._undo)
self.test_all_btn.clicked.connect(self._test_all_servers) self.test_all_btn.clicked.connect(self._test_all_servers)
@@ -1800,6 +2117,7 @@ class MainWindow(QMainWindow):
self.dup_btn, self.dup_btn,
self.del_btn, self.del_btn,
self.paste_btn, self.paste_btn,
self.browse_catalog_btn,
self.copy_btn, self.copy_btn,
self.undo_btn, self.undo_btn,
self.test_all_btn, self.test_all_btn,
@@ -2378,6 +2696,41 @@ class MainWindow(QMainWindow):
self._mark_dirty() self._mark_dirty()
self.status.setText(f"Imported {added} added, {replaced} replaced. Review and Save.") 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): def copy_to_menu(self):
idx = self._current_index() idx = self._current_index()
if not (0 <= idx < len(self.servers)): 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.") QMessageBox.warning(self, "Can't save yet", "Fix the highlighted problem first.")
return return
# Placeholder guard (issue #10): a catalog Add can leave a
# <PLACEHOLDER>-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 "<PLACEHOLDER>" 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 <PLACEHOLDER> 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. # 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 # Compare mtime AND size (not mtime alone) so a concurrent external write
# that lands within the mtime resolution window, or that restores the # that lands within the mtime resolution window, or that restores the
+5 -1
View File
@@ -31,7 +31,11 @@ a = Analysis(
["bcc.py"], ["bcc.py"],
pathex=[], pathex=[],
binaries=[], binaries=[],
datas=[("icons", "icons"), ("data/catalog.json", "data")], datas=[
("icons", "icons"),
("data/catalog.json", "data"),
("data/catalog.json.sig", "data"),
],
hiddenimports=[], hiddenimports=[],
hookspath=[], hookspath=[],
hooksconfig={}, hooksconfig={},
+499 -41
View File
@@ -29,6 +29,7 @@ import tempfile
import threading import threading
import time import time
from dataclasses import dataclass from dataclasses import dataclass
from datetime import date
from pathlib import Path from pathlib import Path
from typing import NamedTuple from typing import NamedTuple
from urllib.parse import urlparse from urllib.parse import urlparse
@@ -2164,6 +2165,30 @@ def restart_claude_desktop() -> RestartResult:
# is rejected by validate_catalog() regardless of how plausible it looks. # is rejected by validate_catalog() regardless of how plausible it looks.
CATALOG_ALLOWED_COMMANDS = frozenset({"npx", "uvx", "docker", "node", "python", "python3"}) CATALOG_ALLOWED_COMMANDS = frozenset({"npx", "uvx", "docker", "node", "python", "python3"})
# Env var keys a catalog entry's config.env must never set. Every one of
# these is a loader/interpreter override that lets a value walk straight
# past CATALOG_ALLOWED_COMMANDS and the -e/--eval/-c deny-rule below: e.g.
# NODE_OPTIONS="--require /tmp/x.js" turns an allowlisted `npx` entry into
# arbitrary code execution without ever touching config.args, which is the
# only field the allowlist/deny-rules/ASCII/secret checks used to cover.
# Matched case-insensitively -- env keys are case-sensitive on POSIX, but a
# `node_options` lookalike is exactly the kind of thing this must catch.
CATALOG_DENIED_ENV_KEYS = frozenset(
{
"NODE_OPTIONS",
"PYTHONSTARTUP",
"PYTHONPATH",
"PYTHONHOME",
"LD_PRELOAD",
"LD_LIBRARY_PATH",
"DYLD_INSERT_LIBRARIES",
"DYLD_LIBRARY_PATH",
"BROWSER",
"PATH",
"NODE_REPL_EXTERNAL_MODULE",
}
)
# Ed25519 public keys allowed to sign a catalog, raw 32-byte form. A LIST # Ed25519 public keys allowed to sign a catalog, raw 32-byte form. A LIST
# (not a single key) so keys can be rotated without bricking installs that # (not a single key) so keys can be rotated without bricking installs that
# still trust an older key: verify_catalog_signature() accepts a match # still trust an older key: verify_catalog_signature() accepts a match
@@ -2189,6 +2214,36 @@ _CATALOG_SECRET_ARG_RE = re.compile(r"(?i)--api[-_]?key=|--token=|--password=")
# <PLACEHOLDER>-style tokens the GUI must have the user fill in before Save. # <PLACEHOLDER>-style tokens the GUI must have the user fill in before Save.
_PLACEHOLDER_RE = re.compile(r"<[^<>\s]+>") _PLACEHOLDER_RE = re.compile(r"<[^<>\s]+>")
# A catalog entry's id becomes an mcpServers JSON key AND is interpolated
# into Qt.AutoText widgets (status bar, QMessageBox) -- an id like
# "<b>Verified</b>" renders as markup there. Not RCE, but UI spoofing, so
# ids are constrained to a plain lowercase slug.
_CATALOG_ID_RE = re.compile(r"^[a-z0-9][a-z0-9._-]{0,63}$")
# Docker flags that consume the next arg as a value (so that value must not
# be mistaken for the image reference when locating it in config.args).
_DOCKER_VALUE_FLAGS = frozenset(
{
"-e",
"--env",
"-v",
"--volume",
"-p",
"--publish",
"--name",
"-w",
"--workdir",
"-u",
"--user",
"--entrypoint",
"--network",
"--platform",
"--add-host",
"-l",
"--label",
}
)
# How many versions a single accepted catalog jump may leap in one go. Bounds # How many versions a single accepted catalog jump may leap in one go. Bounds
# a "freeze" attack: a compromised/leaked signing key claiming an absurd # a "freeze" attack: a compromised/leaked signing key claiming an absurd
# future version would otherwise permanently outrank every legitimate # future version would otherwise permanently outrank every legitimate
@@ -2263,6 +2318,121 @@ def _docker_arg_violations(tag: str, args: list[str]) -> list[str]:
return problems return problems
def _catalog_package_spec_version(spec: str) -> str | None:
"""
Extract the version pin from an npm-style package spec, or None if the
spec carries no pin.
Handles unscoped "name@version" and scoped "@scope/name@version" --
scoped names have a leading "@" that is NOT the version separator, so a
naive split on the first/only "@" misparses "@scope/pkg" (no version)
as pinned to "scope/pkg". Splitting from the right side instead is safe
for both forms because a package name may contain "@" only as the
scope's leading character.
"""
if spec.startswith("@"):
rest = spec[1:]
if "@" not in rest:
return None
_, _, version = rest.rpartition("@")
return version or None
if "@" not in spec:
return None
_, _, version = spec.rpartition("@")
return version or None
def _catalog_package_spec_pinned(spec: str) -> bool:
"""
True if `spec` carries an exact version pin. Covers npm's "name@version"
/ "@scope/name@version" and uv's documented PyPI pin forms
"name@version" and "name==version".
"""
if "==" in spec:
_, _, version = spec.partition("==")
return bool(version)
return bool(_catalog_package_spec_version(spec))
def _first_catalog_package_spec(args: list[str]) -> str | None:
"""
The first arg that could plausibly BE a package spec: skip flags
(leading "-") and <PLACEHOLDER> tokens (which can't be validated and
are filled in by the user later, never shipped by the catalog as the
package name itself). Everything after the first hit is ignored --
trailing flags, paths, and placeholders are not package specs.
"""
for a in args:
if a.startswith("-"):
continue
if _PLACEHOLDER_RE.fullmatch(a):
continue
return a
return None
def _catalog_pin_violations(tag: str, command: str, args: list[str]) -> list[str]:
"""
Version-pinning enforcement (finding #3): a catalog PR can otherwise
ship `npx -y @scope/pkg` or `docker run img:latest` and the *next*
resolve of that package/image is whatever the registry serves that day
-- outside review, outside the signature's meaning. This is the only
place that enforces pinning at runtime; catalog_review.py's
risk_unpinned_package() is a maintainer-facing hint, not a gate.
"""
if command in ("npx", "uvx"):
spec = _first_catalog_package_spec(args)
if spec is None:
return [f"{tag}: config.args must include a package spec to pin (e.g. name@1.2.3)."]
if not _catalog_package_spec_pinned(spec):
return [
f"{tag}: config.args package {spec!r} is not version-pinned; use "
"name@version, @scope/name@version, or name==version."
]
return []
if command == "docker":
image = _docker_image_ref(args)
if image is None:
return [f"{tag}: config.args docker command has no image reference to pin."]
_, sep, image_tag = image.rpartition(":")
if not sep or "/" in image_tag:
return [
f"{tag}: config.args docker image {image!r} has no explicit tag; "
"pin an exact version (not 'latest', not untagged)."
]
if image_tag == "latest":
return [
f"{tag}: config.args docker image {image!r} uses the 'latest' tag, "
"which is not allowed; pin an exact version."
]
return []
return []
def _docker_image_ref(args: list[str]) -> str | None:
"""
Locate the image reference in a `docker run ...` args list: skip the
"run" subcommand and any flags, including ones that consume the next
token as a value (-e, -v, --name, ...) so that value isn't mistaken for
the image. The first remaining positional token is the image.
"""
i = 0
if i < len(args) and args[i] == "run":
i += 1
while i < len(args):
a = args[i]
if a.startswith("-"):
if a in _DOCKER_VALUE_FLAGS and "=" not in a:
i += 2
else:
i += 1
continue
return a
return None
def _validate_catalog_config(tag: str, config) -> list[str]: def _validate_catalog_config(tag: str, config) -> list[str]:
"""Validate the `config` block of a basic-tier catalog entry.""" """Validate the `config` block of a basic-tier catalog entry."""
if not isinstance(config, dict): if not isinstance(config, dict):
@@ -2283,10 +2453,11 @@ def _validate_catalog_config(tag: str, config) -> list[str]:
f"({', '.join(sorted(CATALOG_ALLOWED_COMMANDS))})." f"({', '.join(sorted(CATALOG_ALLOWED_COMMANDS))})."
) )
args = config.get("args") raw_args = config.get("args")
if not isinstance(args, list) or not all(isinstance(a, str) for a in args): args_ok = isinstance(raw_args, list) and all(isinstance(a, str) for a in raw_args)
args = raw_args if args_ok else []
if not args_ok:
problems.append(f"{tag}: config.args must be a list of strings.") problems.append(f"{tag}: config.args must be a list of strings.")
args = []
for a in args: for a in args:
if not a.isascii(): if not a.isascii():
@@ -2305,11 +2476,55 @@ def _validate_catalog_config(tag: str, config) -> list[str]:
if command == "docker": if command == "docker":
problems.extend(_docker_arg_violations(tag, args)) problems.extend(_docker_arg_violations(tag, args))
# Version pinning (finding #3) -- only meaningful once command/args are
# actually well-formed; a malformed args list already got its own
# problem above and has nothing left to pin-check.
if args_ok and command in ("npx", "uvx", "docker"):
problems.extend(_catalog_pin_violations(tag, command, args))
env = config.get("env") env = config.get("env")
if env is not None and ( if env is not None:
not isinstance(env, dict) or any(not isinstance(v, str) for v in env.values()) env_ok = isinstance(env, dict) and all(
): isinstance(k, str) and isinstance(v, str) for k, v in env.items()
)
if not env_ok:
problems.append(f"{tag}: config.env must be an object of string values.") problems.append(f"{tag}: config.env must be an object of string values.")
else:
# config.env (finding #2): unlike args, env was previously
# type-checked ONLY -- no allowlist, no deny-rule, no ASCII
# check, no secret check. That made it the single easiest way
# to smuggle a payload past every other guard in this
# function: an allowlisted `command: npx` plus
# NODE_OPTIONS=--require /tmp/x.js in env walks straight past
# the command allowlist AND the -e/--eval/-c deny-rule above,
# because neither of those ever looks at env.
for key, value in env.items():
if not key.isascii():
problems.append(
f"{tag}: config.env key {key!r} must be ASCII "
"(non-ASCII code points rejected)."
)
if key.upper() in CATALOG_DENIED_ENV_KEYS:
problems.append(
f"{tag}: config.env key {key!r} is on the catalog deny-list "
"(interpreter/loader override) and is not allowed."
)
if not value.isascii():
problems.append(
f"{tag}: config.env value for {key!r} must be ASCII "
"(non-ASCII code points rejected)."
)
if _is_secret_value(value):
problems.append(
f"{tag}: config.env[{key!r}] looks like a real secret value; "
"catalog entries must never ship secret values."
)
if value != "" and not _PLACEHOLDER_RE.fullmatch(value):
problems.append(
f"{tag}: config.env[{key!r}] must be an empty string or a "
"single <PLACEHOLDER> token -- the catalog declares which env "
"vars a server needs, it never supplies their values."
)
return problems return problems
@@ -2329,6 +2544,12 @@ def _validate_catalog_entry(idx: int, entry, seen_ids: set[str]) -> list[str]:
tag = f"servers[{idx}] ({entry_id!r})" tag = f"servers[{idx}] ({entry_id!r})"
if not entry_id.isascii(): if not entry_id.isascii():
problems.append(f"{tag}: 'id' must be ASCII (non-ASCII code points rejected).") problems.append(f"{tag}: 'id' must be ASCII (non-ASCII code points rejected).")
elif not _CATALOG_ID_RE.match(entry_id):
problems.append(
f"{tag}: 'id' must match ^[a-z0-9][a-z0-9._-]{{0,63}}$ "
"(it becomes an mcpServers JSON key and is interpolated into "
"Qt.AutoText widgets)."
)
if entry_id in seen_ids: if entry_id in seen_ids:
problems.append(f"{tag}: duplicate id.") problems.append(f"{tag}: duplicate id.")
seen_ids.add(entry_id) seen_ids.add(entry_id)
@@ -2442,15 +2663,32 @@ def verify_catalog_signature(raw: bytes, sig: bytes, pubkeys: list[bytes]) -> bo
return False return False
def _verify_catalog_candidate(candidate: tuple[bytes, bytes] | None) -> tuple[dict | None, int]:
"""Verify+load+validate one (raw, sig) candidate. Returns (None, -1) on any failure."""
if not candidate:
return None, -1
raw, sig = candidate
if not verify_catalog_signature(raw, sig, CATALOG_PUBKEYS):
return None, -1
try:
data = load_catalog(raw)
except (ValueError, TypeError):
return None, -1
if validate_catalog(data):
return None, -1
return data, catalog_version(data)
def resolve_catalog( def resolve_catalog(
bundled: tuple[bytes, bytes] | None, bundled: tuple[bytes, bytes] | None,
cached: tuple[bytes, bytes] | None, cached: tuple[bytes, bytes] | None,
remote: tuple[bytes, bytes] | None, remote: tuple[bytes, bytes] | None,
floor: int = 0,
) -> dict: ) -> dict:
""" """
Pick the highest-version catalog among bundled/cached/remote. Each Pick the highest-version catalog among bundled/cached/remote. Each of
argument is either None (unavailable) or an (raw_bytes, signature_bytes) bundled/cached/remote is either None (unavailable) or an (raw_bytes,
pair. signature_bytes) pair.
🔴 SECURITY: every candidate — including `bundled`, the copy frozen into 🔴 SECURITY: every candidate — including `bundled`, the copy frozen into
this binary — is verified against CATALOG_PUBKEYS and re-validated from this binary — is verified against CATALOG_PUBKEYS and re-validated from
@@ -2461,46 +2699,68 @@ def resolve_catalog(
by virtue of being local. Signing (and checking the signature at by virtue of being local. Signing (and checking the signature at
runtime, every time) closes that. runtime, every time) closes that.
Anti-rollback: a candidate's version is never accepted if it's lower `floor` is a pure, caller-supplied lower bound (e.g. a persisted
than the best verified candidate already found in this same resolution "last accepted version" the GUI can load from disk and pass in) — this
pass — an attacker replaying an old, since-superseded signed catalog function does no storage of its own.
can't downgrade you.
Anti-freeze: a candidate whose version leaps more than Anti-rollback / anti-freeze, and WHY they apply to every candidate
_CATALOG_MAX_VERSION_JUMP past the current best is also rejected. A including the first one evaluated: the previous version of this
compromised/leaked signing key claiming an absurd future version would function only ran these checks `if best_version >= 0`, i.e. once a
otherwise permanently outrank every legitimate release from then on, candidate had already been accepted in this pass. That let the FIRST
since the resolver always prefers the highest verified version — this verified candidate through unconditionally — a signed catalog claiming
caps how far a single accepted jump can go. version=999999999 sailed straight past both guards if it happened to be
evaluated first, and rollback protection reset on every call anyway
(nothing persisted across restarts). Now both guards are anchored to
something that doesn't depend on iteration order:
- The anti-freeze cap is measured against the BUNDLED catalog's version
(verified independently, once), not against "whatever was accepted
so far in this loop." Bundled ships inside the binary, so it's the
one candidate that isn't attacker-supplied at resolve time — the
natural trust anchor. If bundled itself doesn't verify, `floor` is
the anchor instead.
- The anti-rollback floor is max(floor, bundled's version), so a
caller that persists `floor` across restarts gets real rollback
protection; a caller that doesn't still gets "never below bundled."
On a version TIE, the bundled candidate wins over cached/remote (it
previously lost ties to whichever candidate happened to be evaluated
last, silently preferring remote over bundled at equal version).
Returns the winning catalog dict, or {} if nothing verified and Returns the winning catalog dict, or {} if nothing verified and
validated. validated.
""" """
bundled_data, bundled_version = _verify_catalog_candidate(bundled)
anchor = bundled_version if bundled_version >= 0 else floor
min_accepted = max(floor, bundled_version if bundled_version >= 0 else 0)
candidates = (
("bundled", bundled_data, bundled_version),
("cached", *_verify_catalog_candidate(cached)),
("remote", *_verify_catalog_candidate(remote)),
)
best: dict = {} best: dict = {}
best_version = -1 best_version = -1
best_is_bundled = False
for candidate in (bundled, cached, remote): for source, data, version in candidates:
if not candidate: if data is None:
continue
raw, sig = candidate
if not verify_catalog_signature(raw, sig, CATALOG_PUBKEYS):
continue
try:
data = load_catalog(raw)
except (ValueError, TypeError):
continue
if validate_catalog(data):
continue continue
if version < min_accepted:
continue # anti-rollback / below the persisted floor
if version > anchor + _CATALOG_MAX_VERSION_JUMP:
continue # anti-freeze, capped against the bundled trust anchor
version = catalog_version(data) is_bundled = source == "bundled"
if best_version >= 0: better = version > best_version or (
if version < best_version: version == best_version and is_bundled and not best_is_bundled
continue # anti-rollback )
if version > best_version + _CATALOG_MAX_VERSION_JUMP: if better:
continue # anti-freeze
best = data best = data
best_version = version best_version = version
best_is_bundled = is_bundled
return best return best
@@ -2509,8 +2769,17 @@ def catalog_entry_to_paste_json(entry: dict) -> dict:
""" """
Convert a basic-tier catalog entry into the {name: {command, args, env}} Convert a basic-tier catalog entry into the {name: {command, args, env}}
shape parse_pasted_json()/_import_server() already understand, so the shape parse_pasted_json()/_import_server() already understand, so the
(future) catalog picker dialog can feed a selection straight into the Browse-catalog dialog can feed a selection straight into the existing
existing paste-import path instead of growing a parallel one. 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 {} config = entry.get("config") or {}
name = entry.get("id") or entry.get("display") or "server" name = entry.get("id") or entry.get("display") or "server"
@@ -2518,9 +2787,11 @@ def catalog_entry_to_paste_json(entry: dict) -> dict:
"command": config.get("command", ""), "command": config.get("command", ""),
"args": list(config.get("args") or []), "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: if env:
data["env"] = dict(env) data["env"] = env
return {str(name): data} return {str(name): data}
@@ -2540,3 +2811,190 @@ def config_has_unfilled_placeholders(cfg: dict) -> bool:
if isinstance(env, dict): if isinstance(env, dict):
values.extend(v for v in env.values() if isinstance(v, str)) values.extend(v for v in env.values() if isinstance(v, str))
return any(_PLACEHOLDER_RE.search(v) for v in values) 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 <PLACEHOLDER>-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 []
+656 -4
View File
@@ -1690,8 +1690,12 @@ def _minimal_catalog(version: int = 1) -> dict:
"official": True, "official": True,
"setup": "basic", "setup": "basic",
"config": { "config": {
# Pinned on purpose (issue #68 finding 3): an earlier
# version of this fixture used an unpinned package and
# asserted it validated clean, which enshrined the bug
# instead of catching it.
"command": "npx", "command": "npx",
"args": ["-y", "widget-mcp"], "args": ["-y", "widget-mcp@1.0.0"],
}, },
"placeholders": {}, "placeholders": {},
"env_required": {}, "env_required": {},
@@ -1952,6 +1956,213 @@ def test_validate_catalog_rejects_duplicate_ids():
assert any("duplicate id" in p for p in problems) assert any("duplicate id" in p for p in problems)
# --- validate_catalog: config.env (issue #68 finding 2) -------------------- #
def test_validate_catalog_rejects_each_denied_env_key():
for key in sorted(c.CATALOG_DENIED_ENV_KEYS):
data = _catalog_with(
{"config": {"command": "npx", "args": ["-y", "widget-mcp@1.0.0"], "env": {key: ""}}}
)
problems = c.validate_catalog(data)
assert any("deny-list" in p for p in problems), (key, problems)
def test_validate_catalog_rejects_denied_env_key_case_insensitively():
data = _catalog_with(
{
"config": {
"command": "npx",
"args": ["-y", "widget-mcp@1.0.0"],
"env": {"node_options": ""},
}
}
)
problems = c.validate_catalog(data)
assert any("deny-list" in p for p in problems)
def test_validate_catalog_rejects_nonempty_nonplaceholder_env_value():
data = _catalog_with(
{
"config": {
"command": "npx",
"args": ["-y", "widget-mcp@1.0.0"],
"env": {"FOO_URL": "https://example.com"},
}
}
)
problems = c.validate_catalog(data)
assert any("empty string or a single <PLACEHOLDER>" in p for p in problems)
def test_validate_catalog_accepts_empty_and_placeholder_env_values():
data = _catalog_with(
{
"config": {
"command": "npx",
"args": ["-y", "widget-mcp@1.0.0"],
"env": {"FOO": "", "BAR_URL": "<BAR_URL>"},
}
}
)
assert c.validate_catalog(data) == []
def test_validate_catalog_rejects_non_ascii_env_key():
data = _catalog_with(
{
"config": {
"command": "npx",
"args": ["-y", "widget-mcp@1.0.0"],
"env": {"FÖO": ""},
}
}
)
problems = c.validate_catalog(data)
assert any("config.env key" in p and "ASCII" in p for p in problems)
def test_validate_catalog_rejects_non_ascii_env_value():
data = _catalog_with(
{
"config": {
"command": "npx",
"args": ["-y", "widget-mcp@1.0.0"],
"env": {"FOO": "<Bäd>"},
}
}
)
problems = c.validate_catalog(data)
assert any("config.env value" in p and "ASCII" in p for p in problems)
def test_validate_catalog_rejects_secret_looking_env_value():
data = _catalog_with(
{
"config": {
"command": "npx",
"args": ["-y", "widget-mcp@1.0.0"],
"env": {"SOME_TOKEN": "ghp_abcdef1234567890"},
}
}
)
problems = c.validate_catalog(data)
assert any("real secret value" in p for p in problems)
def test_validate_catalog_rejects_node_options_env_walking_past_allowlist():
# The exact reproduction from issue #68 finding 2: an allowlisted
# `npx` command carrying NODE_OPTIONS in env, which previously passed
# validation and would have flowed straight into the executed
# subprocess via catalog_entry_to_paste_json().
data = _catalog_with(
{
"config": {
"command": "npx",
"args": ["-y", "widget-mcp@1.0.0"],
"env": {"NODE_OPTIONS": "--require /tmp/payload.js"},
}
}
)
problems = c.validate_catalog(data)
assert problems != []
# --- validate_catalog: version pinning (issue #68 finding 3) --------------- #
def test_validate_catalog_rejects_unpinned_npx_package():
data = _catalog_with({"config": {"command": "npx", "args": ["-y", "widget-mcp"]}})
problems = c.validate_catalog(data)
assert any("not version-pinned" in p for p in problems)
def test_validate_catalog_rejects_unpinned_scoped_npx_package():
data = _catalog_with({"config": {"command": "npx", "args": ["-y", "@scope/pkg"]}})
problems = c.validate_catalog(data)
assert any("not version-pinned" in p for p in problems)
def test_validate_catalog_accepts_pinned_scoped_npx_package():
data = _catalog_with({"config": {"command": "npx", "args": ["-y", "@scope/pkg@1.2.3"]}})
assert c.validate_catalog(data) == []
def test_validate_catalog_rejects_unpinned_uvx_package():
data = _catalog_with({"config": {"command": "uvx", "args": ["some-tool"]}})
problems = c.validate_catalog(data)
assert any("not version-pinned" in p for p in problems)
def test_validate_catalog_accepts_uvx_at_version_pin():
data = _catalog_with({"config": {"command": "uvx", "args": ["some-tool@1.0.0"]}})
assert c.validate_catalog(data) == []
def test_validate_catalog_accepts_uvx_double_equals_pin():
data = _catalog_with({"config": {"command": "uvx", "args": ["some-tool==1.0.0"]}})
assert c.validate_catalog(data) == []
def test_validate_catalog_rejects_docker_latest_tag():
data = _catalog_with({"config": {"command": "docker", "args": ["run", "some/image:latest"]}})
problems = c.validate_catalog(data)
assert any("'latest'" in p for p in problems)
def test_validate_catalog_rejects_docker_untagged_image():
data = _catalog_with({"config": {"command": "docker", "args": ["run", "some/image"]}})
problems = c.validate_catalog(data)
assert any("no explicit tag" in p for p in problems)
def test_validate_catalog_accepts_pinned_docker_image_with_flags():
data = _catalog_with(
{
"config": {
"command": "docker",
"args": ["run", "-i", "--rm", "-e", "SOME_TOKEN", "some/image:1.2.3"],
}
}
)
assert c.validate_catalog(data) == []
def test_validate_catalog_does_not_pin_check_placeholders_flags_or_subcommand():
# A pinned uvx spec followed by flags and a <PLACEHOLDER> positional
# must not itself get mistaken for an unpinned package.
data = _catalog_with(
{
"config": {
"command": "uvx",
"args": ["mcp-server-git@2026.7.10", "--repository", "<REPO_PATH>"],
}
}
)
assert c.validate_catalog(data) == []
# --- validate_catalog: id constraint (issue #68 finding 7) ----------------- #
def test_validate_catalog_rejects_id_with_markup():
data = _catalog_with({"id": "<b>Verified</b>"})
problems = c.validate_catalog(data)
assert any("must match" in p for p in problems)
def test_validate_catalog_rejects_id_with_uppercase():
data = _catalog_with({"id": "Widget"})
problems = c.validate_catalog(data)
assert any("must match" in p for p in problems)
def test_validate_catalog_rejects_id_starting_with_dash():
data = _catalog_with({"id": "-widget"})
problems = c.validate_catalog(data)
assert any("must match" in p for p in problems)
def test_validate_catalog_accepts_valid_slug_id():
data = _catalog_with({"id": "widget-2.thing-ok"})
assert c.validate_catalog(data) == []
# --- resolve_catalog -------------------------------------------------------- # # --- resolve_catalog -------------------------------------------------------- #
def test_resolve_catalog_nothing_available_returns_empty_dict(): def test_resolve_catalog_nothing_available_returns_empty_dict():
assert c.resolve_catalog(None, None, None) == {} assert c.resolve_catalog(None, None, None) == {}
@@ -2003,13 +2214,93 @@ def test_resolve_catalog_rejects_absurd_version_jump(monkeypatch):
pub = priv.public_key().public_bytes_raw() pub = priv.public_key().public_bytes_raw()
monkeypatch.setattr(c, "CATALOG_PUBKEYS", [pub]) monkeypatch.setattr(c, "CATALOG_PUBKEYS", [pub])
cached = _signed(_minimal_catalog(version=5), priv) # The freeze attempt goes FIRST (as `cached`), the legitimate catalog
# SECOND (as `remote`) -- on purpose. Putting the good catalog first
# (as an earlier version of this test did) never exercises the
# vulnerable path: the old implementation only guarded a candidate
# against "the best accepted so far," so whichever candidate was
# evaluated FIRST got in unconditionally, uncapped. Ordering the freeze
# attempt first is what actually proves the cap holds regardless of
# evaluation order.
freeze_attempt = _signed(_minimal_catalog(version=999999), priv) freeze_attempt = _signed(_minimal_catalog(version=999999), priv)
good = _signed(_minimal_catalog(version=5), priv)
result = c.resolve_catalog(None, cached, freeze_attempt) result = c.resolve_catalog(None, freeze_attempt, good)
assert c.catalog_version(result) == 5 assert c.catalog_version(result) == 5
def test_resolve_catalog_caps_first_and_only_candidate(monkeypatch):
# issue #68 finding 6: with no bundled catalog to anchor against, a
# signed catalog claiming an absurd version must still be capped even
# when it is the ONLY candidate resolve_catalog() ever sees -- there is
# no "best so far" for it to be compared against, so the cap has to
# apply unconditionally, not "once something else has already landed."
priv = Ed25519PrivateKey.generate()
pub = priv.public_key().public_bytes_raw()
monkeypatch.setattr(c, "CATALOG_PUBKEYS", [pub])
freeze_attempt = _signed(_minimal_catalog(version=999999999), priv)
result = c.resolve_catalog(None, None, freeze_attempt)
assert result == {}
def test_resolve_catalog_anchors_cap_to_bundled_not_a_chained_best(monkeypatch):
# Anti-freeze must be measured against the BUNDLED version specifically,
# not against "whatever the best-so-far happens to be after each
# candidate is accepted" -- a chained anchor lets each accepted
# candidate ratchet the allowed ceiling upward, so a legitimate
# moderate bump (cached) plus a second, much larger jump (remote) can
# each individually look "within _CATALOG_MAX_VERSION_JUMP of the
# previous one" while remote is nowhere near bundled's version.
priv = Ed25519PrivateKey.generate()
pub = priv.public_key().public_bytes_raw()
monkeypatch.setattr(c, "CATALOG_PUBKEYS", [pub])
bundled = _signed(_minimal_catalog(version=2), priv)
cached = _signed(_minimal_catalog(version=1000), priv) # within 1000 of bundled
remote = _signed(_minimal_catalog(version=1900), priv) # within 1000 of cached,
# NOT of bundled
result = c.resolve_catalog(bundled, cached, remote)
assert c.catalog_version(result) == 1000
def test_resolve_catalog_prefers_bundled_on_version_tie(monkeypatch):
# issue #68 finding 6: on a tie the LAST candidate evaluated used to
# win, so remote silently beat bundled at equal version. Bundled --
# the copy frozen into the binary -- must win ties.
priv = Ed25519PrivateKey.generate()
pub = priv.public_key().public_bytes_raw()
monkeypatch.setattr(c, "CATALOG_PUBKEYS", [pub])
bundled_data = _minimal_catalog(version=5)
bundled = _signed(bundled_data, priv)
remote_data = _minimal_catalog(version=5)
remote_data["servers"][0]["display"] = "Remote Impostor"
remote = _signed(remote_data, priv)
result = c.resolve_catalog(bundled, None, remote)
assert c.catalog_version(result) == 5
assert result["servers"][0]["display"] == "Widget"
def test_resolve_catalog_floor_rejects_below_persisted_version(monkeypatch):
# `floor` is a pure parameter: the caller (eventually the GUI, from
# persisted storage) can pass a previously-accepted version, and
# nothing below it may be accepted even with no bundled catalog to
# anchor against.
priv = Ed25519PrivateKey.generate()
pub = priv.public_key().public_bytes_raw()
monkeypatch.setattr(c, "CATALOG_PUBKEYS", [pub])
stale = _signed(_minimal_catalog(version=3), priv)
result = c.resolve_catalog(None, None, stale, floor=10)
assert result == {}
def test_resolve_catalog_malformed_candidate_does_not_raise(monkeypatch): def test_resolve_catalog_malformed_candidate_does_not_raise(monkeypatch):
priv = Ed25519PrivateKey.generate() priv = Ed25519PrivateKey.generate()
pub = priv.public_key().public_bytes_raw() pub = priv.public_key().public_bytes_raw()
@@ -2039,15 +2330,69 @@ def test_resolve_catalog_invalid_but_signed_candidate_is_skipped(monkeypatch):
def test_catalog_entry_to_paste_json_basic_shape(): def test_catalog_entry_to_paste_json_basic_shape():
entry = _minimal_catalog()["servers"][0] entry = _minimal_catalog()["servers"][0]
result = c.catalog_entry_to_paste_json(entry) result = c.catalog_entry_to_paste_json(entry)
assert result == {"widget": {"command": "npx", "args": ["-y", "widget-mcp"]}} assert result == {"widget": {"command": "npx", "args": ["-y", "widget-mcp@1.0.0"]}}
def test_catalog_entry_to_paste_json_includes_env_when_present(): def test_catalog_entry_to_paste_json_includes_env_when_present():
# NOTE: catalog_entry_to_paste_json() is a pure shape-converter for an
# entry that has ALREADY passed validate_catalog() -- it is correct for
# it to carry env through verbatim. The bug (issue #68 finding 2) was
# never in this function; it was that validate_catalog() let entries
# with dangerous/non-placeholder env values reach this function in the
# first place. This test now proves that boundary explicitly: a
# validation-legal env value (a <PLACEHOLDER> token) survives the
# conversion, and a value validate_catalog() would have rejected is
# confirmed rejected before it ever gets here.
entry = _minimal_catalog()["servers"][0] entry = _minimal_catalog()["servers"][0]
entry["config"]["env"] = {"GRAFANA_URL": "<GRAFANA_URL>"} entry["config"]["env"] = {"GRAFANA_URL": "<GRAFANA_URL>"}
result = c.catalog_entry_to_paste_json(entry) result = c.catalog_entry_to_paste_json(entry)
assert result["widget"]["env"] == {"GRAFANA_URL": "<GRAFANA_URL>"} assert result["widget"]["env"] == {"GRAFANA_URL": "<GRAFANA_URL>"}
catalog = _minimal_catalog()
catalog["servers"][0]["config"]["env"] = {"GRAFANA_URL": "<GRAFANA_URL>"}
assert c.validate_catalog(catalog) == []
malicious = _minimal_catalog()
malicious["servers"][0]["config"]["env"] = {"NODE_OPTIONS": "--require /tmp/payload.js"}
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": "<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_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(): def test_config_has_unfilled_placeholders_true_for_token():
cfg = {"command": "npx", "args": ["-y", "server", "<ALLOWED_DIR>"]} cfg = {"command": "npx", "args": ["-y", "server", "<ALLOWED_DIR>"]}
@@ -2062,3 +2407,310 @@ def test_config_has_unfilled_placeholders_false_after_fill():
def test_config_has_unfilled_placeholders_checks_env_too(): def test_config_has_unfilled_placeholders_checks_env_too():
cfg = {"command": "uvx", "args": ["mcp-grafana"], "env": {"GRAFANA_URL": "<GRAFANA_URL>"}} cfg = {"command": "uvx", "args": ["mcp-grafana"], "env": {"GRAFANA_URL": "<GRAFANA_URL>"}}
assert c.config_has_unfilled_placeholders(cfg) is True 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", "<ALLOWED_DIR>"],
"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"}