2 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
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
11 changed files with 1320 additions and 1748 deletions
+1 -1
View File
@@ -120,7 +120,7 @@ jobs:
# below to the new key's base64 form, as its own reviewed change. # 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: env:
EXPECTED_CATALOG_PUBKEY_B64: "0s24PmkZcTT5yxNDdyTPHl5fyxArrHNPJKBjnXoQd8k=" EXPECTED_CATALOG_PUBKEY_B64: "082NOwVB7uURkvfyS3+knJ+40Fk6C9unsF47+2uPKo4="
run: | run: |
python - <<'PY' python - <<'PY'
import base64, os, pathlib, sys import base64, os, pathlib, sys
+3 -35
View File
@@ -90,21 +90,6 @@ key, because they protect different things and live in different places:
| Generated with | `python catalog_console.py keygen` | `python catalog_console.py keygen --release` | | Generated with | `python catalog_console.py keygen` | `python catalog_console.py keygen --release` |
| Exported for CI with | *(never — there is no supported way to export this key)* | `python catalog_console.py show-seed-b64 --release` | | Exported for CI with | *(never — there is no supported way to export this key)* | `python catalog_console.py show-seed-b64 --release` |
**Confused about which key is which, or what state either is in?** Run:
```bash
python catalog_console.py keys
```
It needs no passphrase (it never touches private key bytes) and prints a
plain-English report for both keys: where each private half lives, whether
it's present on this machine, its fingerprint, whether that fingerprint
matches what's actually committed in `bcc_core.py`, `ci.yml`'s trust
anchor, and `scripts/sign_checksums.py`, and whether
`data/catalog.json.sig` currently verifies — ending with the exact command
to run next for whatever state it finds. This is the check that would have
caught [issue #68](../../issues/68)'s finding 5 incident before it happened.
**Why two keys:** the catalog key is the root of trust for what BCC **Why two keys:** the catalog key is the root of trust for what BCC
actually *executes* on a user's machine — every `command`/`args` pair in actually *executes* on a user's machine — every `command`/`args` pair in
the shipped catalog is only there because this key signed it. If that key the shipped catalog is only there because this key signed it. If that key
@@ -125,29 +110,12 @@ refuses to run without `--release` specifically so the catalog seed can't
be exported by habit or muscle memory. be exported by habit or muscle memory.
**Release signing public key** (Ed25519, base64, raw 32 bytes) — this is **Release signing public key** (Ed25519, base64, raw 32 bytes) — this is
the RELEASE key, which signs `SHA256SUMS` (release checksums). It does the RELEASE key, not the catalog key:
**not** sign `data/catalog.json` and is not the key `bcc_core.CATALOG_PUBKEYS`
trusts:
``` ```
6BnPgJEHJFyVltFoLTCNadIsehjy00iiW8IRlC1TfhA= <PLACEHOLDER — AJ: paste the release public key from `catalog_console.py keygen --release` here>
``` ```
The catalog public key (Ed25519, base64, raw 32 bytes) — this is the key
that signs `data/catalog.json` and is trusted via `bcc_core.CATALOG_PUBKEYS`
and the CI trust anchor in `.github/workflows/ci.yml`. It is listed here
for completeness, not because you need it to verify a download — use the
*release* key above for that:
```
0s24PmkZcTT5yxNDdyTPHl5fyxArrHNPJKBjnXoQd8k=
```
Both keys above were rotated 2026-07 — see [issue #68](../../issues/68)
finding 5. The prior (shared) key is retired and is deliberately **not**
kept in either trust list; retaining a burned key would defeat the point
of rotating it.
## Run from source ## Run from source
```bash ```bash
@@ -222,7 +190,7 @@ file is also listed, marked *legacy*, so you can copy them over.
- `bcc.spec` — PyInstaller build spec (cross-platform). - `bcc.spec` — PyInstaller build spec (cross-platform).
- `scripts/build_icons.py` — regenerates `icons/app.icns` and `icons/app.ico` from source PNGs. - `scripts/build_icons.py` — regenerates `icons/app.icns` and `icons/app.ico` from source PNGs.
- `scripts/sign_checksums.py` — generates and Ed25519-signs the release `SHA256SUMS` manifest (see [Verifying your download](#verifying-your-download)). - `scripts/sign_checksums.py` — generates and Ed25519-signs the release `SHA256SUMS` manifest (see [Verifying your download](#verifying-your-download)).
- `catalog_console.py` / `catalog_review.py`**maintainer-only**, never shipped to users (excluded from `bcc.spec`; see `tests/test_catalog_console_packaging.py`). The Catalog Console: review + sign `data/catalog.json` (against `main`, an open PR, or the branch you have checked out — `--ref <branch>` to be explicit, e.g. mid key-rotation, so a rotation can be signed and pushed to its own branch *before* it's merged, never forcing a red `main`), generate/manage both signing keys (`keygen`, `keygen --release`), and report on their status (`keys`, no passphrase needed) — see [Signing keys](#signing-keys). - `catalog_console.py` / `catalog_review.py`**maintainer-only**, never shipped to users (excluded from `bcc.spec`; see `tests/test_catalog_console_packaging.py`). The Catalog Console: review + sign `data/catalog.json`, and generate/manage both signing keys (`keygen`, `keygen --release`) — see [Signing keys](#signing-keys).
## Building from source ## Building from source
+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={},
+204 -5
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
@@ -2193,7 +2194,7 @@ CATALOG_DENIED_ENV_KEYS = frozenset(
# still trust an older key: verify_catalog_signature() accepts a match # still trust an older key: verify_catalog_signature() accepts a match
# against ANY key in this list. # against ANY key in this list.
CATALOG_PUBKEYS: list[bytes] = [ CATALOG_PUBKEYS: list[bytes] = [
base64.b64decode("0s24PmkZcTT5yxNDdyTPHl5fyxArrHNPJKBjnXoQd8k="), base64.b64decode("082NOwVB7uURkvfyS3+knJ+40Fk6C9unsF47+2uPKo4="),
] ]
# Domain-separation prefix for the signed message. The signature covers # Domain-separation prefix for the signed message. The signature covers
@@ -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}} 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"
@@ -2777,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}
@@ -2799,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 []
+64 -553
View File
@@ -10,20 +10,14 @@ for data/catalog.json (issue #62).
Flow: Load -> Review -> Sign. Flow: Load -> Review -> Sign.
1. Load -- pick a source: an open Gitea PR touching data/catalog.json, 1. Load -- pick a source: an open Gitea PR touching data/catalog.json,
the current tip of `main`, or the branch this checkout is or the current tip of `main`. The Console fetches the exact
currently ON (or --ref names) -- the latter exists so a git blob (via a local clone's git plumbing) and PINS its
catalog-signing-key rotation, or any other catalog change blob SHA *and the ref it came from* for the rest of this
landed on a branch, can be reviewed and SIGNED before that review pass. For a PR, the diff is against `main`; for
branch is ever merged, instead of merging a PR that leaves `main`, the diff is against the last catalog a maintainer
main red and fixing it up afterwards (issue #68). The actually SIGNED (the bytes covered by the current
Console fetches the exact git blob (via a local clone's git data/catalog.json.sig), never against itself -- an empty
plumbing) and PINS its blob SHA *and the ref it came from* diff must mean "nothing to sign", never "sign unlocked".
for the rest of this review pass. For a PR, the diff is
against `main`; for `main` or a branch, the diff is against
the last catalog a maintainer actually SIGNED on that same
ref (the bytes covered by the current data/catalog.json.sig
there), never against itself -- an empty diff must mean
"nothing to sign", never "sign unlocked".
2. Review -- a semantic diff (catalog_review.diff_catalogs), one card per 2. Review -- a semantic diff (catalog_review.diff_catalogs), one card per
changed entry, with risk annotations changed entry, with risk annotations
(catalog_review.entry_risk_findings). A registry lookup for (catalog_review.entry_risk_findings). A registry lookup for
@@ -43,23 +37,13 @@ Flow: Load -> Review -> Sign.
blocking risk finding (catalog_review.sign_precondition / blocking risk finding (catalog_review.sign_precondition /
can_sign -- the TOCTOU fix). On success, writes can_sign -- the TOCTOU fix). On success, writes
data/catalog.json + data/catalog.json.sig and commits BOTH data/catalog.json + data/catalog.json.sig and commits BOTH
in a single commit, then pushes to the SAME ref that was in a single commit, then pushes -- so main is never red
reviewed (never a hardcoded "main") -- so that ref is never between a catalog merge and its signature. Signing uses the
red between a catalog merge and its signature, whether
that ref is `main` or a rotation branch. Signing uses the
CATALOG key ONLY -- see "Two signing keys" below. CATALOG key ONLY -- see "Two signing keys" below.
The signature must be the artefact of an actual review, not a step that The signature must be the artefact of an actual review, not a step that
follows one. Signing IS the approval act. follows one. Signing IS the approval act.
Confused about which key is which, or what state either is in? Run
`python catalog_console.py keys` -- it needs no passphrase and prints a
plain-English status report for both keys: where each private half lives,
whether it's present, its fingerprint, whether that fingerprint matches
what's committed in bcc_core.py / ci.yml / scripts/sign_checksums.py, and
whether data/catalog.json.sig currently verifies -- ending with exactly
what to run next.
Two signing keys (issue #68 finding 5): the CATALOG key (offline, Two signing keys (issue #68 finding 5): the CATALOG key (offline,
Console-only, `keygen` / `keygen --release` picks which) is the root of Console-only, `keygen` / `keygen --release` picks which) is the root of
trust for what BCC executes and must never touch CI. The RELEASE key is trust for what BCC executes and must never touch CI. The RELEASE key is
@@ -85,7 +69,6 @@ import urllib.error
import urllib.request import urllib.request
from dataclasses import dataclass from dataclasses import dataclass
from pathlib import Path from pathlib import Path
from typing import ClassVar
import bcc_core as core import bcc_core as core
import catalog_review as review import catalog_review as review
@@ -214,89 +197,6 @@ def unlock_signing_key(passphrase: str, kind: str = "catalog") -> bytes:
return review.decrypt_private_key(blob, passphrase) return review.decrypt_private_key(blob, passphrase)
def _pubkey_cache_file(kind: str) -> Path:
assert kind in _KEY_KINDS, f"unknown key kind {kind!r}, expected one of {_KEY_KINDS}"
return KEY_STORAGE_DIR / f"signing_key_{kind}.pub"
def store_public_key(pubkey: bytes, kind: str = "catalog") -> None:
"""Cache the PUBLIC half of a signing key, in plain base64, next to its
encrypted private counterpart (OS keychain if available, else the file
cache). Public keys are not secret -- this cache exists purely so
`catalog_console.py keys` can report a fingerprint and compare it
against what's committed in source WITHOUT ever decrypting (or asking
for a passphrase to unlock) the private key. Called by cmd_keygen()
right after a keypair is generated.
"""
keyring = _keyring_module()
pub_b64 = base64.b64encode(pubkey).decode("ascii")
if keyring is not None:
try:
keyring.set_password(_KEYRING_SERVICE, f"{_keyring_username(kind)}-pub", pub_b64)
return
except Exception:
pass # fall through to the file-based cache
KEY_STORAGE_DIR.mkdir(parents=True, exist_ok=True)
_pubkey_cache_file(kind).write_text(pub_b64 + "\n", encoding="utf-8")
def load_public_key(kind: str = "catalog") -> bytes | None:
"""The cached PUBLIC key of the given `kind`, or None if no key of that
kind has been generated yet (or it predates public-key caching -- an
older keygen run that never called store_public_key). Never touches the
encrypted private blob and never asks for a passphrase."""
keyring = _keyring_module()
if keyring is not None:
try:
pub_b64 = keyring.get_password(_KEYRING_SERVICE, f"{_keyring_username(kind)}-pub")
if pub_b64:
return base64.b64decode(pub_b64)
except Exception:
pass
path = _pubkey_cache_file(kind)
if not path.exists():
return None
text = path.read_text(encoding="utf-8").strip()
if not text:
return None
try:
return base64.b64decode(text)
except ValueError:
return None
def has_encrypted_key(kind: str = "catalog") -> bool:
"""Whether a private key of this `kind` has been generated on this
machine -- checked WITHOUT decrypting or asking for a passphrase, so
`catalog_console.py keys` can report existence unconditionally."""
keyring = _keyring_module()
if keyring is not None:
try:
if keyring.get_password(_KEYRING_SERVICE, _keyring_username(kind)):
return True
except Exception:
pass
return _key_storage_file(kind).exists()
def describe_local_key_location(kind: str) -> str:
"""Human-readable description of where a `kind` key's PRIVATE half
lives on this machine, for the `keys` report and the sign-flow prompt.
Never the release key's CI location -- that's cmd_keys' job to append,
since it's true regardless of whether a local copy also exists."""
keyring = _keyring_module()
if keyring is not None:
try:
if keyring.get_password(_KEYRING_SERVICE, _keyring_username(kind)):
return "on this machine, in the OS keychain (encrypted, passphrase-protected)"
except Exception:
pass
key_file = _key_storage_file(kind)
if key_file.exists():
return f"on this machine, at {key_file} (encrypted, passphrase-protected)"
return "not generated yet"
# --------------------------------------------------------------------------- # # --------------------------------------------------------------------------- #
# git plumbing against a local clone. The clone's `origin` remote is assumed # git plumbing against a local clone. The clone's `origin` remote is assumed
# to already carry credentials (the "tokened remote" every other BCC # to already carry credentials (the "tokened remote" every other BCC
@@ -323,40 +223,6 @@ def fetch_ref(repo_dir: Path, ref: str) -> str:
return _git(repo_dir, "rev-parse", "FETCH_HEAD").strip() return _git(repo_dir, "rev-parse", "FETCH_HEAD").strip()
def current_branch(repo_dir: Path) -> str | None:
"""The branch currently checked out at `repo_dir`, or None if it can't
be determined (detached HEAD, bare repo, mid-rebase, ...). This is what
lets the Console offer "sign against the branch I already have checked
out" as a source (issue #68 rotation-completability fix) without the
maintainer having to type the branch name -- ReviewWindow defaults to
it unless --ref names one explicitly.
"""
try:
name = _git(repo_dir, "rev-parse", "--abbrev-ref", "HEAD").strip()
except GitError:
return None
return None if name in ("", "HEAD") else name
def compute_own_refs(explicit_ref: str | None, detected_branch: str | None) -> list[str]:
""" "main" plus (if different) `explicit_ref` or `detected_branch` -- the
exact list of refs ReviewWindow offers as "own" sources (loadable,
signable, AND pushable directly, unlike a PR's read-only
refs/pull/<n>/head). Pure: `detected_branch` is injected (normally
current_branch(repo_dir)) so this ref-resolution seam -- the fix for
issue #68's "rotation can't complete without a red main" -- is testable
without git or Qt. "main" is always index 0 so a stale/absent
list-widget selection still defaults sanely (see ReviewWindow._on_load's
row-clamping, and cmd_gui/ReviewWindow.__init__ for how `explicit_ref`
is threaded from --ref).
"""
refs = ["main"]
branch = explicit_ref or detected_branch
if branch and branch not in refs:
refs.append(branch)
return refs
def blob_sha_at(repo_dir: Path, commit: str, path: str) -> str: def blob_sha_at(repo_dir: Path, commit: str, path: str) -> str:
"""The git blob SHA of `path` as it exists at `commit`. This is what """The git blob SHA of `path` as it exists at `commit`. This is what
gets pinned at review-start and re-checked immediately before signing gets pinned at review-start and re-checked immediately before signing
@@ -417,48 +283,17 @@ def last_signed_catalog_raw(repo_dir: Path, commit: str) -> bytes | None:
return review.find_last_signed_catalog_raw(candidates, sig_bytes, core.CATALOG_PUBKEYS) return review.find_last_signed_catalog_raw(candidates, sig_bytes, core.CATALOG_PUBKEYS)
def catalog_signature_valid_at(repo_dir: Path, commit: str, raw: bytes) -> bool:
"""Whether data/catalog.json.sig as of `commit` verifies `raw` against
the CURRENTLY-TRUSTED bcc_core.CATALOG_PUBKEYS.
False whenever there is no .sig file, or the .sig exists but was
produced by a key that is not (or no longer) in CATALOG_PUBKEYS -- most
notably right after a catalog signing-key rotation (issue #68 finding 5
follow-up), when the committed .sig was produced by the now-retired old
key. This is the rotation-detection primitive ReviewWindow._on_load
uses to decide whether to enter re-attestation mode
(review.start_review(..., reattest=True)) instead of the ordinary
diff-against-last-signed path.
Delegates to bcc_core.verify_catalog_signature -- never reimplemented,
per this module's docstring ("one source of truth").
"""
try:
sig_sha = blob_sha_at(repo_dir, commit, SIG_PATH)
except GitError:
return False
sig_bytes = blob_bytes(repo_dir, sig_sha)
return core.verify_catalog_signature(raw, sig_bytes, core.CATALOG_PUBKEYS)
def commit_and_push_signed_catalog( def commit_and_push_signed_catalog(
repo_dir: Path, raw_bytes: bytes, signature: bytes, *, branch: str = "main" repo_dir: Path, raw_bytes: bytes, signature: bytes, *, branch: str = "main"
) -> str: ) -> str:
"""Write data/catalog.json + data/catalog.json.sig and commit BOTH in a """Write data/catalog.json + data/catalog.json.sig and commit BOTH in a
single commit, then push to `branch`. Returns the new commit SHA. single commit, then push to `branch`. Returns the new commit SHA.
`branch` MUST be the same ref that was actually reviewed
(ReviewWindow._on_sign passes `session.loaded_ref`, never a hardcoded
"main" -- issue #68 completability fix): a catalog change reviewed on a
branch has to be signed and pushed to THAT branch so the branch itself
is never red, rather than landing the signature on main after a merge.
This is deliberate: if signing happened in a commit AFTER the catalog This is deliberate: if signing happened in a commit AFTER the catalog
merge, `branch` would be red (payload present, signature missing) merge, main would be red (payload present, signature missing) between
between the catalog merge and its signing commit. Routine red branches every catalog merge and its signing commit. Routine red-main trains
train exactly the alarm fatigue this whole design exists to prevent. exactly the alarm fatigue this whole design exists to prevent. Emitting
Emitting one commit with both files means `branch` is never in that one commit with both files means main is never in that state.
state -- whether `branch` is main or a rotation-in-progress branch.
""" """
_git(repo_dir, "checkout", branch) _git(repo_dir, "checkout", branch)
_git(repo_dir, "pull", "--ff-only", "origin", branch) _git(repo_dir, "pull", "--ff-only", "origin", branch)
@@ -648,76 +483,14 @@ def registry_fetcher(ref: review.PackageRef) -> dict | None:
return None return None
# --------------------------------------------------------------------------- #
# Key status ("keys" command, issue #62/#68 follow-up: "make key handling
# comprehensible"). File I/O only -- the actual status/report logic is pure
# and lives in catalog_review (key_status / render_key_status_report), so
# it's unit-testable without touching disk. These wrappers read the THREE
# places a signing key's public half is expected to be committed, from
# `repo_dir`'s working tree, so `keys` reports on whatever ref/branch is
# actually checked out there (never this process's own sys.path import).
# --------------------------------------------------------------------------- #
def read_catalog_pubkeys_from_source(repo_dir: Path) -> list[bytes]:
path = repo_dir / "bcc_core.py"
if not path.exists():
return []
return review.extract_pubkey_list_literal(path.read_text(encoding="utf-8"), "CATALOG_PUBKEYS")
def read_release_pubkeys_from_source(repo_dir: Path) -> list[bytes]:
path = repo_dir / "scripts" / "sign_checksums.py"
if not path.exists():
return []
return review.extract_pubkey_list_literal(path.read_text(encoding="utf-8"), "RELEASE_PUBKEYS")
def read_ci_trust_anchor_pubkeys(repo_dir: Path) -> list[bytes]:
path = repo_dir / ".github" / "workflows" / "ci.yml"
if not path.exists():
return []
pubkey = review.extract_ci_trust_anchor_pubkey(path.read_text(encoding="utf-8"))
return [pubkey] if pubkey is not None else []
def catalog_sig_status_on_disk(repo_dir: Path, committed_catalog_pubkeys: list[bytes]) -> str:
""" "valid" / "invalid" / "missing" for data/catalog.json.sig as it sits
in `repo_dir`'s WORKING TREE right now (not a git ref -- the `keys`
command deliberately reports on-disk state, which is what "the current
checked-out branch" concretely means and avoids a network fetch just to
print a status line). Verified against `committed_catalog_pubkeys`
(whatever bcc_core.py on this same working tree currently says), not
the running process's own bcc_core import, so this stays correct no
matter which ref/branch happens to be checked out.
"""
catalog_path = repo_dir / CATALOG_PATH
sig_path = repo_dir / SIG_PATH
if not catalog_path.exists() or not sig_path.exists():
return "missing"
raw = catalog_path.read_bytes()
sig = sig_path.read_bytes()
if core.verify_catalog_signature(raw, sig, committed_catalog_pubkeys):
return "valid"
return "invalid"
# --------------------------------------------------------------------------- # # --------------------------------------------------------------------------- #
# GUI (PySide6). Everything above this line has no Qt dependency and is # GUI (PySide6). Everything above this line has no Qt dependency and is
# exercised by tests/test_catalog_review.py; everything below is a thin # exercised by tests/test_catalog_review.py; everything below is a thin
# shell that calls into it. # shell that calls into it.
# --------------------------------------------------------------------------- # # --------------------------------------------------------------------------- #
# PySide6 is imported defensively: `keygen`, `show-seed-b64`, and `keys` from PySide6.QtCore import Qt, QThread, Signal # noqa: E402
# (the commands a maintainer runs most often, and the ones this file's from PySide6.QtWidgets import ( # noqa: E402
# tests/py_compile-only CI environment can exercise) have no GUI dependency
# at all and must keep working even somewhere PySide6 isn't installed or
# won't import (e.g. no system Qt libs). Only `gui` needs it -- cmd_gui()
# checks _PYSIDE6_AVAILABLE and fails with a clear message instead of an
# ImportError stack trace if it's missing.
try:
from PySide6.QtCore import Qt, QThread, Signal
from PySide6.QtWidgets import (
QApplication, QApplication,
QCheckBox, QCheckBox,
QDialog, QDialog,
@@ -736,15 +509,7 @@ try:
QVBoxLayout, QVBoxLayout,
QWidget, QWidget,
) )
except ImportError as _pyside6_exc: # pragma: no cover - only hit where PySide6 is absent
_PYSIDE6_IMPORT_ERROR: str | None = str(_pyside6_exc)
else:
_PYSIDE6_IMPORT_ERROR = None
_PYSIDE6_AVAILABLE = _PYSIDE6_IMPORT_ERROR is None
if _PYSIDE6_AVAILABLE: # pragma: no branch - GUI class defs, only skipped where PySide6 is absent
def plain_label(text: object) -> QLabel: def plain_label(text: object) -> QLabel:
"""A QLabel guaranteed to render `text` as plain text, never HTML. """A QLabel guaranteed to render `text` as plain text, never HTML.
@@ -762,8 +527,10 @@ if _PYSIDE6_AVAILABLE: # pragma: no branch - GUI class defs, only skipped where
label.setWordWrap(True) label.setWordWrap(True)
return label return label
_SEVERITY_PREFIX = {"blocking": "✖ BLOCKING", "warning": "⚠ WARNING", "info": " INFO"} _SEVERITY_PREFIX = {"blocking": "✖ BLOCKING", "warning": "⚠ WARNING", "info": " INFO"}
class RegistryLookupWorker(QThread): class RegistryLookupWorker(QThread):
"""Off-UI-thread registry lookups, mirroring bcc.py's ConnTester/ """Off-UI-thread registry lookups, mirroring bcc.py's ConnTester/
SpawnTester pattern. Never blocks the review UI on a slow/dead network.""" SpawnTester pattern. Never blocks the review UI on a slow/dead network."""
@@ -782,6 +549,7 @@ if _PYSIDE6_AVAILABLE: # pragma: no branch - GUI class defs, only skipped where
] ]
self.done.emit(results) self.done.emit(results)
class EntryCard(QWidget): class EntryCard(QWidget):
"""One changed catalog entry: the diff, risk findings, and the """One changed catalog entry: the diff, risk findings, and the
acknowledge checkbox that gates Sign. `command`/`args` are rendered acknowledge checkbox that gates Sign. `command`/`args` are rendered
@@ -893,24 +661,15 @@ if _PYSIDE6_AVAILABLE: # pragma: no branch - GUI class defs, only skipped where
self.registry_label.setText(html.escape(text)) self.registry_label.setText(html.escape(text))
self.registry_label.setTextFormat(Qt.PlainText) self.registry_label.setTextFormat(Qt.PlainText)
class PassphraseDialog(QDialog): class PassphraseDialog(QDialog):
"""Prompts for a signing key's passphrase. def __init__(self, prompt: str, parent=None):
`prompt` must say plainly WHICH key (CATALOG or RELEASE) is about to be
unlocked, and its fingerprint when known -- issue #62/#68 follow-up: a
maintainer must see which key he's about to type a passphrase for
BEFORE typing it, not infer it from context. This is the exact ambiguity
that led to a private key being pasted into a chat window.
"""
def __init__(self, prompt: str, window_title: str = "Signing key passphrase", parent=None):
super().__init__(parent) super().__init__(parent)
self.setWindowTitle(window_title) self.setWindowTitle("Signing key passphrase")
layout = QFormLayout(self) layout = QFormLayout(self)
layout.addRow(plain_label(prompt))
self.edit = QLineEdit() self.edit = QLineEdit()
self.edit.setEchoMode(QLineEdit.EchoMode.Password) self.edit.setEchoMode(QLineEdit.EchoMode.Password)
layout.addRow("Passphrase:", self.edit) layout.addRow(prompt, self.edit)
buttons = QDialogButtonBox( buttons = QDialogButtonBox(
QDialogButtonBox.StandardButton.Ok | QDialogButtonBox.StandardButton.Cancel QDialogButtonBox.StandardButton.Ok | QDialogButtonBox.StandardButton.Cancel
) )
@@ -921,26 +680,14 @@ if _PYSIDE6_AVAILABLE: # pragma: no branch - GUI class defs, only skipped where
def passphrase(self) -> str: def passphrase(self) -> str:
return self.edit.text() return self.edit.text()
class ReviewWindow(QMainWindow):
#: Placeholder empty catalog used when there is nothing to diff against
#: yet (no prior signature, or a key rotation invalidated the old one).
_EMPTY_CATALOG: ClassVar[dict] = {"schema": 1, "version": 0, "servers": []}
def __init__(self, repo_dir: Path, ref: str | None = None): class ReviewWindow(QMainWindow):
def __init__(self, repo_dir: Path):
super().__init__() super().__init__()
self.repo_dir = repo_dir self.repo_dir = repo_dir
self.session: review.ReviewSession | None = None self.session: review.ReviewSession | None = None
self.cards: dict[str, EntryCard] = {} self.cards: dict[str, EntryCard] = {}
# "Own" refs this clone can load/diff/sign+push against directly --
# issue #68 rotation-completability fix. "main" is always offered;
# if the checkout is on a different branch (or --ref names one
# explicitly), that branch is offered too, so a rotation in
# progress on a branch can be reviewed, signed, and pushed to ITS
# OWN ref -- completing the rotation before merge, never forcing a
# red main in between. See _load_own_ref() / _on_sign().
self._own_refs = self._compute_own_refs(ref)
self.setWindowTitle("BCC Catalog Console -- maintainer-only, never shipped") self.setWindowTitle("BCC Catalog Console -- maintainer-only, never shipped")
central = QWidget() central = QWidget()
self.setCentralWidget(central) self.setCentralWidget(central)
@@ -948,13 +695,7 @@ if _PYSIDE6_AVAILABLE: # pragma: no branch - GUI class defs, only skipped where
top = QHBoxLayout() top = QHBoxLayout()
self.source_list = QListWidget() self.source_list = QListWidget()
for own_ref in self._own_refs: self.source_list.addItem(QListWidgetItem("main (current tip)"))
label = (
f"{own_ref} (current tip)"
if own_ref == "main"
else f"{own_ref} (current branch)"
)
self.source_list.addItem(QListWidgetItem(label))
top.addWidget(self.source_list, 1) top.addWidget(self.source_list, 1)
side = QVBoxLayout() side = QVBoxLayout()
@@ -968,22 +709,6 @@ if _PYSIDE6_AVAILABLE: # pragma: no branch - GUI class defs, only skipped where
top.addLayout(side) top.addLayout(side)
root.addLayout(top) root.addLayout(top)
# KEY-ROTATION RE-ATTESTATION BANNER (issue #68 finding 5 follow-up).
# Hidden until _on_load() detects that the loaded catalog's
# signature does NOT verify under the currently-trusted
# bcc_core.CATALOG_PUBKEYS (catalog_signature_valid_at()) -- most
# commonly, right after the maintainer rotates the catalog signing
# key. This mode must NEVER be entered silently: this banner is the
# only thing standing between "every entry needs re-review" and a
# maintainer wondering why the diff view suddenly shows 19
# "added" entries with no explanation.
self.reattest_banner = plain_label("")
self.reattest_banner.setStyleSheet(
"background-color: #c62828; color: white; font-weight: bold; padding: 8px;"
)
self.reattest_banner.setVisible(False)
root.addWidget(self.reattest_banner)
self.scroll = QScrollArea() self.scroll = QScrollArea()
self.scroll.setWidgetResizable(True) self.scroll.setWidgetResizable(True)
self.card_container = QWidget() self.card_container = QWidget()
@@ -1003,66 +728,41 @@ if _PYSIDE6_AVAILABLE: # pragma: no branch - GUI class defs, only skipped where
self._prs: list[CatalogPR] = [] self._prs: list[CatalogPR] = []
self._refresh_pr_list() self._refresh_pr_list()
def _compute_own_refs(self, explicit_ref: str | None) -> list[str]:
"""Thin GUI-side wrapper: injects current_branch(self.repo_dir)
(a git call) into the pure compute_own_refs() -- see that
function's docstring for what this list actually means."""
return compute_own_refs(explicit_ref, current_branch(self.repo_dir))
def _refresh_pr_list(self): def _refresh_pr_list(self):
self._prs = list_open_catalog_prs(self._token) self._prs = list_open_catalog_prs(self._token)
while self.source_list.count() > len(self._own_refs): while self.source_list.count() > 1:
self.source_list.takeItem(len(self._own_refs)) self.source_list.takeItem(1)
for pr in self._prs: for pr in self._prs:
self.source_list.addItem(QListWidgetItem(f"PR #{pr.number}: {pr.title}")) self.source_list.addItem(QListWidgetItem(f"PR #{pr.number}: {pr.title}"))
def _load_own_ref(self, loaded_ref: str): def _on_load(self):
"""Load + diff `loaded_ref` -- "main" or the maintainer's own row = self.source_list.currentRow()
working branch, ANY ref this clone's origin can fetch directly (as try:
opposed to a PR's read-only refs/pull/<n>/head). Diffs against the if row <= 0:
last catalog a maintainer actually SIGNED on that ref (never # source = main: diff against the last catalog a maintainer
against itself -- an empty diff must mean "nothing to sign", never # actually SIGNED (the bytes covered by the current
"sign unlocked", issue #68 finding 1) and enters KEY-ROTATION # data/catalog.json.sig), never against itself. Diffing
RE-ATTESTATION mode if the committed .sig doesn't verify under the # main-vs-main is what made an empty diff -> instantly
currently-trusted bcc_core.CATALOG_PUBKEYS (issue #68 finding 5 # "signable" in the first place (issue #68 finding 1) --
follow-up). This is the exact logic "main" always used -- pulled # this is the only path that reaches sign_catalog_bytes(),
into its own method, parameterized on the ref, so a rotation on a # so if it can't be trusted nothing can.
branch gets the SAME treatment as one on main and can be signed loaded_ref = "main"
(and pushed back to that SAME branch, see _on_sign) before merge --
the completability fix this method exists for.
Returns (new_raw, new_blob_sha, new_catalog, old_catalog, reattest).
"""
commit = fetch_ref(self.repo_dir, loaded_ref) commit = fetch_ref(self.repo_dir, loaded_ref)
new_raw, new_blob_sha = read_catalog_at_commit(self.repo_dir, commit) new_raw, new_blob_sha = read_catalog_at_commit(self.repo_dir, commit)
new_catalog = core.load_catalog(new_raw) new_catalog = core.load_catalog(new_raw)
if not catalog_signature_valid_at(self.repo_dir, commit, new_raw):
reattest = True
old_catalog = dict(self._EMPTY_CATALOG)
else:
reattest = False
old_raw = last_signed_catalog_raw(self.repo_dir, commit) old_raw = last_signed_catalog_raw(self.repo_dir, commit)
old_catalog = ( old_catalog = (
core.load_catalog(old_raw) if old_raw is not None else dict(self._EMPTY_CATALOG) core.load_catalog(old_raw)
) if old_raw is not None
else {
return new_raw, new_blob_sha, new_catalog, old_catalog, reattest "schema": 1,
"version": 0,
def _on_load(self): "servers": [],
row = self.source_list.currentRow() }
if row < 0:
row = 0 # nothing explicitly selected -- default to index 0 ("main")
try:
if row < len(self._own_refs):
loaded_ref = self._own_refs[row]
new_raw, new_blob_sha, new_catalog, old_catalog, reattest = self._load_own_ref(
loaded_ref
) )
else: else:
pr = self._prs[row - len(self._own_refs)] pr = self._prs[row - 1]
loaded_ref = pr.head_ref loaded_ref = pr.head_ref
reattest = False
commit = fetch_ref(self.repo_dir, loaded_ref) commit = fetch_ref(self.repo_dir, loaded_ref)
old_commit = fetch_ref(self.repo_dir, "main") old_commit = fetch_ref(self.repo_dir, "main")
@@ -1081,21 +781,8 @@ if _PYSIDE6_AVAILABLE: # pragma: no branch - GUI class defs, only skipped where
# from the SAME ref that was reviewed, instead of a hardcoded # from the SAME ref that was reviewed, instead of a hardcoded
# "main" -- see review.sign_precondition and issue #68 finding 1. # "main" -- see review.sign_precondition and issue #68 finding 1.
self.session = review.start_review( self.session = review.start_review(
new_blob_sha, old_catalog, new_catalog, loaded_ref=loaded_ref, reattest=reattest new_blob_sha, old_catalog, new_catalog, loaded_ref=loaded_ref
) )
if reattest:
self.reattest_banner.setText(
"KEY ROTATION IN PROGRESS -- the trusted CATALOG signing key changed. "
"The existing data/catalog.json.sig does NOT verify under the current "
"bcc_core.CATALOG_PUBKEYS, so it is NOT trusted. Every one of the "
f"{len(self.session.changes)} entries below must be re-reviewed and "
"acknowledged before the new CATALOG key can re-sign this catalog -- "
"this is the intended cost of rotating the CATALOG key, not a bug."
)
self.reattest_banner.setVisible(True)
else:
self.reattest_banner.setVisible(False)
self.reattest_banner.setText("")
self._render_cards() self._render_cards()
def _render_cards(self): def _render_cards(self):
@@ -1107,11 +794,7 @@ if _PYSIDE6_AVAILABLE: # pragma: no branch - GUI class defs, only skipped where
assert self.session is not None assert self.session is not None
all_ids = sorted( all_ids = sorted(
{ {e.get("id") for e in (self.session.new_catalog.get("servers") or []) if e.get("id")}
e.get("id")
for e in (self.session.new_catalog.get("servers") or [])
if e.get("id")
}
) )
for change in self.session.changes: for change in self.session.changes:
card = EntryCard(change, all_ids) card = EntryCard(change, all_ids)
@@ -1134,12 +817,6 @@ if _PYSIDE6_AVAILABLE: # pragma: no branch - GUI class defs, only skipped where
all_ack = review.all_entries_acknowledged(self.session) all_ack = review.all_entries_acknowledged(self.session)
self.sign_btn.setEnabled(all_ack) self.sign_btn.setEnabled(all_ack)
pending = len(self.session.changes) - len(self.session.acknowledged) pending = len(self.session.changes) - len(self.session.acknowledged)
if self.session.reattest:
self.status_label.setText(
f"RE-ATTESTATION under the new key: {len(self.session.changes)} entries, "
f"{pending} not yet acknowledged."
)
else:
self.status_label.setText( self.status_label.setText(
f"{len(self.session.changes)} changed entries, {pending} not yet acknowledged." f"{len(self.session.changes)} changed entries, {pending} not yet acknowledged."
) )
@@ -1182,52 +859,23 @@ if _PYSIDE6_AVAILABLE: # pragma: no branch - GUI class defs, only skipped where
self._on_load() self._on_load()
return return
# Show WHICH key is about to be used, and its fingerprint, BEFORE dialog = PassphraseDialog("Enter CATALOG signing key passphrase:", self)
# the passphrase field even appears -- issue #62/#68 follow-up. A
# maintainer must never have to infer which key a bare "Enter
# passphrase" prompt means; that ambiguity is exactly what led to a
# private key being pasted into a chat window.
catalog_pubkey = load_public_key("catalog")
if catalog_pubkey is not None:
fp_note = f"fingerprint {review.fingerprint_pubkey(catalog_pubkey)}"
else:
fp_note = "fingerprint unknown (generated before fingerprint caching -- re-run keygen to cache it)"
prompt = (
f"About to sign with the CATALOG signing key ({fp_note}).\n"
"This is the root of trust for what BCC executes on a user's machine -- "
"it is never the RELEASE key.\n\n"
"Enter the CATALOG key's passphrase:"
)
dialog = PassphraseDialog(prompt, "CATALOG signing key passphrase", self)
if dialog.exec() != QDialog.DialogCode.Accepted: if dialog.exec() != QDialog.DialogCode.Accepted:
return return
try: try:
seed = unlock_signing_key(dialog.passphrase(), kind="catalog") seed = unlock_signing_key(dialog.passphrase())
except (FileNotFoundError, ValueError) as e: except (FileNotFoundError, ValueError) as e:
QMessageBox.critical(self, "Sign failed", html.escape(str(e))) QMessageBox.critical(self, "Sign failed", html.escape(str(e)))
return return
signature = review.sign_catalog_bytes(self._new_raw, seed) signature = review.sign_catalog_bytes(self._new_raw, seed)
try: try:
# Push to the SAME ref that was reviewed (session.loaded_ref), new_commit = commit_and_push_signed_catalog(self.repo_dir, self._new_raw, signature)
# never a hardcoded "main" -- issue #68 completability fix. A
# rotation (or any other catalog change) reviewed on a branch
# must land on THAT branch so it can be signed and pushed
# before the branch is ever merged, instead of forcing a merge
# of a red PR followed by a fix-up on main.
new_commit = commit_and_push_signed_catalog(
self.repo_dir, self._new_raw, signature, branch=self.session.loaded_ref
)
except GitError as e: except GitError as e:
QMessageBox.critical(self, "Commit/push failed", html.escape(str(e))) QMessageBox.critical(self, "Commit/push failed", html.escape(str(e)))
return return
QMessageBox.information( QMessageBox.information(self, "Signed", f"Signed and pushed as commit {new_commit[:12]}.")
self,
"Signed",
f"Signed with the CATALOG key and pushed to {self.session.loaded_ref} "
f"as commit {new_commit[:12]}.",
)
self.sign_btn.setEnabled(False) self.sign_btn.setEnabled(False)
@@ -1257,10 +905,9 @@ def cmd_keygen(args: argparse.Namespace) -> int:
blob = review.encrypt_private_key(seed, passphrase) blob = review.encrypt_private_key(seed, passphrase)
where = store_encrypted_key(blob, kind=kind) where = store_encrypted_key(blob, kind=kind)
store_public_key(pubkey, kind=kind) # non-secret; lets `keys` fingerprint without a passphrase
pubkey_b64 = base64.b64encode(pubkey).decode("ascii") pubkey_b64 = base64.b64encode(pubkey).decode("ascii")
print(f"{kind.upper()} private key encrypted and stored in: {where}") print(f"{kind.capitalize()} private key encrypted and stored in: {where}")
print() print()
if kind == "catalog": if kind == "catalog":
print( print(
@@ -1273,12 +920,8 @@ def cmd_keygen(args: argparse.Namespace) -> int:
) )
print() print()
print( print(
"PUBLIC key (base64, safe to commit/share) -- hand this to whoever " "Public key (base64) -- hand this to whoever maintains bcc_core.py so "
"maintains bcc_core.py so they can add it to CATALOG_PUBKEYS (this tool " "they can add it to CATALOG_PUBKEYS (this tool does not edit that file):"
"does not edit that file). There is no PRIVATE-key output for the catalog "
"key: it is never meant to leave this machine, and this tool has no "
"command that exports it (show-seed-b64 refuses without --release, and "
"even then only exports the release key)."
) )
print(f" {pubkey_b64}") print(f" {pubkey_b64}")
else: else:
@@ -1289,23 +932,14 @@ def cmd_keygen(args: argparse.Namespace) -> int:
"contain it)." "contain it)."
) )
print() print()
print( print("1. Public key (base64) -- paste into scripts/sign_checksums.py RELEASE_PUBKEYS:")
"1. PUBLIC key (base64, safe to commit/share) -- paste into "
"scripts/sign_checksums.py RELEASE_PUBKEYS:"
)
print(f" {pubkey_b64}") print(f" {pubkey_b64}")
print() print()
print( print(
"2. PRIVATE key (secret -- NEVER commit, NEVER paste into chat/email) -- " "2. Private key -- add it as the Gitea repo secret RELEASE_SIGNING_KEY "
"add it as the Gitea repo secret RELEASE_SIGNING_KEY (base64 of the " "(base64 of the 32-byte private seed). Get that value with:"
"32-byte private seed). This command does not print it; fetch it "
"separately, when you're ready to paste it straight into the Gitea "
"secret field, with:"
)
print(
" python catalog_console.py show-seed-b64 --release "
"# prints the PRIVATE key -- read the warning banner it shows"
) )
print(" python catalog_console.py show-seed-b64 --release # prints the raw key")
return 0 return 0
@@ -1325,49 +959,17 @@ def cmd_show_seed_b64(args: argparse.Namespace) -> int:
file=sys.stderr, file=sys.stderr,
) )
return 1 return 1
# The prior wording ("Release signing key passphrase:") was ambiguous passphrase = getpass.getpass("Release signing key passphrase: ")
# enough that a maintainer who ran this command, saw that prompt, and
# then saw a base64 blob printed with zero surrounding context, pasted
# the output into a chat believing it was the PUBLIC key. It is not --
# it is the raw private seed. Every string this command prints from here
# down exists to make that mistake structurally harder to make again.
passphrase = getpass.getpass(
"Passphrase for the release signing key (the one YOU chose when generating it): "
)
try: try:
seed = unlock_signing_key(passphrase, kind="release") seed = unlock_signing_key(passphrase, kind="release")
except (FileNotFoundError, ValueError) as e: except (FileNotFoundError, ValueError) as e:
print(f"error: {e}", file=sys.stderr) print(f"error: {e}", file=sys.stderr)
return 1 return 1
# Loud banner on STDERR, seed alone on STDOUT -- so `show-seed-b64
# --release | pbcopy` (or piping into the Gitea secret field) still
# gets ONLY the seed, while anyone watching the terminal still sees the
# warning. Never merge these into one stream.
print("!!! PRIVATE KEY BELOW -- this is the RELEASE_SIGNING_KEY secret value.", file=sys.stderr)
print(
"!!! Paste it ONLY into the Gitea secret field. Never into chat, email, "
"a file, or a commit.",
file=sys.stderr,
)
print(
"!!! Anyone holding this value can forge release checksum signatures.",
file=sys.stderr,
)
print(base64.b64encode(seed).decode("ascii")) print(base64.b64encode(seed).decode("ascii"))
return 0 return 0
def cmd_gui(args: argparse.Namespace) -> int: def cmd_gui(args: argparse.Namespace) -> int:
if not _PYSIDE6_AVAILABLE:
print(
"error: PySide6 is not available in this Python environment, so the GUI "
f"can't launch ({_PYSIDE6_IMPORT_ERROR}). `keygen`, `show-seed-b64`, and "
"`keys` don't need it and still work here.",
file=sys.stderr,
)
return 1
repo_dir = Path(args.repo).resolve() repo_dir = Path(args.repo).resolve()
if not (repo_dir / CATALOG_PATH).exists(): if not (repo_dir / CATALOG_PATH).exists():
print( print(
@@ -1378,98 +980,18 @@ def cmd_gui(args: argparse.Namespace) -> int:
app = QApplication(sys.argv) app = QApplication(sys.argv)
app.setApplicationName("BCC Catalog Console") app.setApplicationName("BCC Catalog Console")
win = ReviewWindow(repo_dir, ref=args.ref) win = ReviewWindow(repo_dir)
win.resize(900, 700) win.resize(900, 700)
win.show() win.show()
return app.exec() return app.exec()
def cmd_keys(args: argparse.Namespace) -> int:
"""`python catalog_console.py keys` -- "which key is what, and what
state is everything in?" (issue #62/#68 follow-up). Reads the CURRENT
WORKING TREE at --repo (so it reports on whatever branch/ref is
actually checked out there -- issue #68 rotation-completability fix
means that's often not "main" during a rotation), gathers every input
the pure catalog_review.key_status()/render_key_status_report() need,
and prints the result. Never touches, decrypts, or prints a private key
-- everything gathered here is a cached PUBLIC key, a fingerprint, file
text, or a signature verification result.
"""
repo_dir = Path(args.repo).resolve()
branch = current_branch(repo_dir) or "(unknown -- detached HEAD or not a git checkout)"
print(f"Catalog Console -- key status for {repo_dir}")
print(f"Checked-out ref: {branch}")
print()
committed_catalog_pubkeys = read_catalog_pubkeys_from_source(repo_dir)
ci_trust_anchor_pubkeys = read_ci_trust_anchor_pubkeys(repo_dir)
committed_release_pubkeys = read_release_pubkeys_from_source(repo_dir)
catalog_status = review.key_status(
"catalog",
display_name="CATALOG",
purpose=(
"Signs the server list that BCC writes into your Claude config. This is "
"what decides which programs run on a user's machine -- the root of trust."
),
private_key_location=describe_local_key_location("catalog"),
local_exists=has_encrypted_key("catalog"),
local_pubkey=load_public_key("catalog"),
locations=[
("bcc_core.CATALOG_PUBKEYS", committed_catalog_pubkeys),
("ci.yml trust anchor (EXPECTED_CATALOG_PUBKEY_B64)", ci_trust_anchor_pubkeys),
],
catalog_sig_status=catalog_sig_status_on_disk(repo_dir, committed_catalog_pubkeys),
)
release_where = describe_local_key_location("release")
if release_where == "not generated yet":
release_where = (
"not generated yet, and not verifiable from here as present in CI "
"(check the Gitea repo secret RELEASE_SIGNING_KEY directly)"
)
else:
release_where = (
f"{release_where} -- intended to be pasted into the Gitea secret "
"RELEASE_SIGNING_KEY (via `show-seed-b64 --release`) and not kept as the "
"primary copy once that's done"
)
release_status = review.key_status(
"release",
display_name="RELEASE",
purpose=(
"Signs the SHA256SUMS checksum manifest for release downloads only. "
"CI-resident on purpose: a CI compromise burns this key, never the "
"catalog key -- that asymmetry is the whole point of having two keys."
),
private_key_location=release_where,
local_exists=has_encrypted_key("release"),
local_pubkey=load_public_key("release"),
locations=[("scripts/sign_checksums.py RELEASE_PUBKEYS", committed_release_pubkeys)],
catalog_sig_status=None,
)
print(review.render_key_status_report([catalog_status, release_status]))
return 0
def build_parser() -> argparse.ArgumentParser: def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description=__doc__) parser = argparse.ArgumentParser(description=__doc__)
sub = parser.add_subparsers(dest="command") sub = parser.add_subparsers(dest="command")
p_gui = sub.add_parser("gui", help="launch the review/sign GUI (default)") p_gui = sub.add_parser("gui", help="launch the review/sign GUI (default)")
p_gui.add_argument("--repo", default=".", help="path to a BCC git checkout (default: cwd)") p_gui.add_argument("--repo", default=".", help="path to a BCC git checkout (default: cwd)")
p_gui.add_argument(
"--ref",
default=None,
help=(
"branch to offer as an additional load/sign/push source, alongside "
"'main' and any open PRs -- e.g. a key-rotation branch, so rotation "
"can be reviewed and signed BEFORE merge instead of forcing a red main "
"(issue #68). Defaults to whatever branch --repo is currently checked "
"out on; only needed if that's not the branch you mean."
),
)
p_gui.set_defaults(func=cmd_gui) p_gui.set_defaults(func=cmd_gui)
p_keygen = sub.add_parser( p_keygen = sub.add_parser(
@@ -1497,21 +1019,10 @@ def build_parser() -> argparse.ArgumentParser:
) )
p_seed.set_defaults(func=cmd_show_seed_b64) p_seed.set_defaults(func=cmd_show_seed_b64)
p_keys = sub.add_parser(
"keys",
help=(
"print a plain-English status report: which key is what, where its "
"private half lives, whether it matches what's committed, and whether "
"data/catalog.json.sig currently verifies"
),
)
p_keys.add_argument("--repo", default=".", help="path to a BCC git checkout (default: cwd)")
p_keys.set_defaults(func=cmd_keys)
return parser return parser
_SUBCOMMANDS = ("gui", "keygen", "show-seed-b64", "keys", "-h", "--help") _SUBCOMMANDS = ("gui", "keygen", "show-seed-b64", "-h", "--help")
def main(argv: list[str] | None = None) -> int: def main(argv: list[str] | None = None) -> int:
-277
View File
@@ -16,8 +16,6 @@ new surface" recurring-bug lesson).
from __future__ import annotations from __future__ import annotations
import base64
import hashlib
import os import os
import re import re
from collections.abc import Callable from collections.abc import Callable
@@ -444,42 +442,17 @@ class ReviewSession:
from the reviewed ref on every PR review (the bug that made the PR path from the reviewed ref on every PR review (the bug that made the PR path
unable to sign at all, and forced everyone onto the vacuous unable to sign at all, and forced everyone onto the vacuous
main-vs-itself path instead). main-vs-itself path instead).
`reattest` marks a KEY-ROTATION re-attestation pass (issue #68 finding 5
follow-up): the currently-trusted `bcc_core.CATALOG_PUBKEYS` key changed
and the existing `data/catalog.json.sig` no longer verifies under it.
Content-wise nothing may have changed -- `diff_catalogs(old, new)` can be
genuinely empty -- but the NEW key has never vouched for any of this
catalog before, so every entry needs a first-time attestation under the
new key, not a diff against the old one. When `reattest` is set,
`changes` is built as "every entry in `new_catalog`, presented as if
newly added" (via `diff_catalogs(None, new_catalog)`) instead of a
diff against `old_catalog`, so the acknowledge-gate in `can_sign()`
requires re-reviewing everything the new key will sign -- which is the
intended cost of a key rotation, not a bypass of the empty-diff guard.
""" """
pinned_blob_sha: str pinned_blob_sha: str
old_catalog: dict old_catalog: dict
new_catalog: dict new_catalog: dict
loaded_ref: str = "main" loaded_ref: str = "main"
reattest: bool = False
changes: list[EntryChange] = field(default_factory=list) changes: list[EntryChange] = field(default_factory=list)
acknowledged: set[str] = field(default_factory=set) acknowledged: set[str] = field(default_factory=set)
def __post_init__(self) -> None: def __post_init__(self) -> None:
if not self.changes: if not self.changes:
if self.reattest:
# Every entry in the new catalog is treated as though it
# were newly added -- because, under the NEW signing key,
# it is: nothing this key signs was ever attested by it
# before. Reusing diff_catalogs(None, new) (rather than a
# bespoke code path) means the same "added" risk predicate
# (risk_new_entry) and the same EntryChange shape the rest
# of this module and the GUI already know how to render
# apply here unmodified.
self.changes = diff_catalogs(None, self.new_catalog)
else:
self.changes = diff_catalogs(self.old_catalog, self.new_catalog) self.changes = diff_catalogs(self.old_catalog, self.new_catalog)
@@ -488,14 +461,12 @@ def start_review(
old_catalog: dict, old_catalog: dict,
new_catalog: dict, new_catalog: dict,
loaded_ref: str = "main", loaded_ref: str = "main",
reattest: bool = False,
) -> ReviewSession: ) -> ReviewSession:
return ReviewSession( return ReviewSession(
pinned_blob_sha=pinned_blob_sha, pinned_blob_sha=pinned_blob_sha,
old_catalog=old_catalog, old_catalog=old_catalog,
new_catalog=new_catalog, new_catalog=new_catalog,
loaded_ref=loaded_ref, loaded_ref=loaded_ref,
reattest=reattest,
) )
@@ -560,14 +531,6 @@ def can_sign(session: ReviewSession, current_blob_sha: str) -> SignDecision:
mean "nothing to sign", never "sign unlocked". (Issue #68 finding 1; mean "nothing to sign", never "sign unlocked". (Issue #68 finding 1;
this is what let commit b08cf21 sign all 19 entries with zero of them this is what let commit b08cf21 sign all 19 entries with zero of them
ever reviewed.) ever reviewed.)
This gate is unaffected by `session.reattest`: a key-rotation
re-attestation session's `changes` is built from
`diff_catalogs(None, new_catalog)` (see ReviewSession), which is
empty ONLY if the catalog itself has zero entries -- a genuinely
empty catalog either way. Rotation never manufactures a non-empty
changeset out of an empty one; it just changes *what* "non-empty"
is computed against.
1. TOCTOU: `current_blob_sha` (fetched fresh, immediately before signing, 1. TOCTOU: `current_blob_sha` (fetched fresh, immediately before signing,
from the ref that was actually reviewed -- see sign_precondition()) from the ref that was actually reviewed -- see sign_precondition())
must match the blob SHA pinned when review began. If the bytes on the must match the blob SHA pinned when review began. If the bytes on the
@@ -853,243 +816,3 @@ _NON_ASCII_RE = re.compile(r"[^\x00-\x7f]")
def contains_non_ascii(s: str) -> bool: def contains_non_ascii(s: str) -> bool:
return bool(_NON_ASCII_RE.search(s)) return bool(_NON_ASCII_RE.search(s))
# --------------------------------------------------------------------------- #
# Key status reporting (issue #62/#68 follow-up: "make key handling
# comprehensible"). Pure functions only -- `catalog_console.py cmd_keys` is a
# thin printer that gathers inputs (local key caches, source-file text, the
# catalog + its .sig) and hands them here. NEVER touches private key bytes:
# every input/output here is a public key, a fingerprint, or a status string.
# --------------------------------------------------------------------------- #
def fingerprint_pubkey(pubkey: bytes) -> str:
"""Short, human-comparable fingerprint of a raw Ed25519 public key: the
first 16 hex chars of its SHA-256 digest, grouped in 4s (e.g. "3F2A 9C1B
44DE 08AA") so two fingerprints can be eyeballed for a mismatch the way a
PGP fingerprint is. Deliberately NOT the raw base64 pubkey itself in the
default short form (that's available via the full committed value in the
report) -- a fixed-width grouped hex string is easier to compare at a
glance and to read aloud/type over chat if needed. Never derived from,
and never printed alongside, any private key material.
"""
digest = hashlib.sha256(pubkey).hexdigest().upper()[:16]
return " ".join(digest[i : i + 4] for i in range(0, len(digest), 4))
_PUBKEY_LIST_B64_RE = re.compile(r'base64\.b64decode\(\s*"([^"]+)"\s*\)')
def extract_pubkey_list_literal(source_text: str, var_name: str) -> list[bytes]:
"""Best-effort extraction of a `<var_name>: list[bytes] = [...]` literal
(each entry a `base64.b64decode("...")` call, matching the exact style
bcc_core.CATALOG_PUBKEYS and scripts.sign_checksums.RELEASE_PUBKEYS are
both written in) straight out of Python source TEXT.
Deliberately a regex over text, not an import: `catalog_console.py keys`
must report on whatever ref/branch is checked out at the inspected repo
path, which may not be (and need not be) importable from the running
process's own sys.path. Returns [] if the variable isn't found in this
exact shape -- callers treat that as "nothing committed here", not an
error, since a report that can't parse a file should say so plainly
rather than crash the whole `keys` command over one malformed file.
"""
match = re.search(
rf"{re.escape(var_name)}\s*:\s*list\[bytes\]\s*=\s*\[(.*?)\]", source_text, re.DOTALL
)
if not match:
return []
keys: list[bytes] = []
for b64 in _PUBKEY_LIST_B64_RE.findall(match.group(1)):
try:
keys.append(base64.b64decode(b64))
except ValueError:
continue
return keys
_CI_TRUST_ANCHOR_RE = re.compile(r'EXPECTED_CATALOG_PUBKEY_B64:\s*"([^"]+)"')
def extract_ci_trust_anchor_pubkey(ci_yml_text: str) -> bytes | None:
"""Best-effort extraction of ci.yml's `EXPECTED_CATALOG_PUBKEY_B64` trust
anchor (issue #68 finding 4) from the workflow file's TEXT. Returns None
if the constant isn't found -- the `keys` report shows that plainly
("not found in ci.yml") rather than raising.
"""
match = _CI_TRUST_ANCHOR_RE.search(ci_yml_text)
if not match:
return None
try:
return base64.b64decode(match.group(1))
except ValueError:
return None
@dataclass(frozen=True)
class PubkeyLocationCheck:
"""One place in the source tree a key's public half is expected to be
committed, and whether the fingerprint(s) found there match the key
stored locally."""
location: str
committed_fingerprints: tuple[str, ...]
status: str # "match" | "mismatch" | "unknown" (no local key to compare against)
@dataclass(frozen=True)
class KeyStatus:
"""Everything `catalog_console.py keys` reports about ONE signing key.
Built by key_status() below; rendered by render_key_status_report().
Never carries private key material -- every field here is safe to print.
"""
kind: str # "catalog" | "release"
display_name: str # "CATALOG" | "RELEASE"
purpose: str # one-line plain-English purpose
private_key_location: str # human-readable, e.g. "on this machine, in the OS keychain"
local_exists: bool
local_fingerprint: str | None
locations: tuple[PubkeyLocationCheck, ...]
catalog_sig_status: str | None = None # "valid" | "invalid" | "missing" | None (n/a)
def key_status(
kind: str,
*,
display_name: str,
purpose: str,
private_key_location: str,
local_exists: bool,
local_pubkey: bytes | None,
locations: list[tuple[str, list[bytes]]],
catalog_sig_status: str | None = None,
) -> KeyStatus:
"""Pure assembly of a KeyStatus from already-resolved inputs (no file or
git I/O here -- that's catalog_console.py's job). `locations` is a list
of (label, committed_pubkeys) pairs, e.g.
[("bcc_core.CATALOG_PUBKEYS", [...]), ("ci.yml trust anchor", [...])],
so a key can be checked against every place its public half is expected
to be committed, independently -- this is the check that would have
caught bcc_core.CATALOG_PUBKEYS and ci.yml's trust anchor silently
drifting apart (issue #68 finding 4 was exactly that kind of drift).
"""
checks: list[PubkeyLocationCheck] = []
for label, committed_pubkeys in locations:
fps = tuple(fingerprint_pubkey(pk) for pk in committed_pubkeys)
if local_pubkey is None:
status = "unknown"
elif local_pubkey in committed_pubkeys:
status = "match"
else:
status = "mismatch"
checks.append(
PubkeyLocationCheck(location=label, committed_fingerprints=fps, status=status)
)
return KeyStatus(
kind=kind,
display_name=display_name,
purpose=purpose,
private_key_location=private_key_location,
local_exists=local_exists,
local_fingerprint=fingerprint_pubkey(local_pubkey) if local_pubkey is not None else None,
locations=tuple(checks),
catalog_sig_status=catalog_sig_status,
)
_LOCATION_STATUS_ICON = {"match": "", "mismatch": "", "unknown": "⚠️"}
_LOCATION_STATUS_VERDICT = {
"match": "MATCHES the local private key",
"mismatch": "DOES NOT MATCH the local private key",
"unknown": "cannot compare -- no local key to check against",
}
_CATALOG_SIG_STATUS_LINE = {
"valid": "✅ data/catalog.json.sig verifies under the committed CATALOG_PUBKEYS.",
"invalid": (
"❌ data/catalog.json.sig does NOT verify under the committed CATALOG_PUBKEYS -- "
"the catalog needs re-signing (Load → acknowledge all → Sign)."
),
"missing": (
"⚠️ data/catalog.json.sig is missing entirely -- the catalog has never been signed."
),
}
def recommend_next_steps(statuses: list[KeyStatus]) -> list[str]:
"""The pure "what to do next" logic behind the keys report's closing
section -- one concrete, runnable-looking instruction per problem found,
naming the exact key involved (never just "the key"). Returns a single
reassuring line if nothing needs attention."""
steps: list[str] = []
for s in statuses:
if not s.local_exists:
flag = " --release" if s.kind == "release" else ""
steps.append(
f"{s.display_name} key has never been generated on this machine -- run "
f"`python catalog_console.py keygen{flag}`."
)
continue
for loc in s.locations:
if loc.status == "mismatch":
steps.append(
f"{s.display_name} key's local fingerprint does not match "
f"{loc.location} -- update {loc.location} to the fingerprint shown "
"above (or, if this is unexpected, treat the committed key as "
"untrusted and investigate before doing anything else)."
)
elif loc.status == "unknown":
steps.append(
f"{s.display_name} key's local fingerprint could not be checked against "
f"{loc.location} -- re-run keygen (or, for an older install, unlock the "
"key once) so its public half is cached locally."
)
if s.kind == "catalog" and s.catalog_sig_status in ("invalid", "missing"):
steps.append(
"The catalog needs re-signing: run `python catalog_console.py gui --repo .` "
"and Load → acknowledge every entry → Sign. If main is red because "
"of a key rotation, load the branch with the rotation instead of main "
"(current-branch / --ref source) so the fix lands before merge."
)
if not steps:
steps.append("Everything is consistent -- no action needed.")
return steps
def render_key_status_report(statuses: list[KeyStatus]) -> str:
"""Render a full, plain-English-first key status report as one string.
`catalog_console.py cmd_keys` prints this verbatim -- the CLI is a thin
printer over this pure function, which is what makes the report's
content (not just its plumbing) unit-testable."""
lines: list[str] = []
for s in statuses:
lines.append(f"=== {s.display_name} KEY ===")
lines.append(s.purpose)
lines.append(f"Private half lives: {s.private_key_location}")
if s.local_exists and s.local_fingerprint:
lines.append(f"Exists locally: yes (fingerprint {s.local_fingerprint})")
elif s.local_exists:
lines.append("Exists locally: yes (fingerprint unknown -- re-run keygen to cache it)")
else:
lines.append("Exists locally: no")
for loc in s.locations:
icon = _LOCATION_STATUS_ICON.get(loc.status, "?")
fps = (
", ".join(loc.committed_fingerprints)
if loc.committed_fingerprints
else "(nothing committed here)"
)
verdict = _LOCATION_STATUS_VERDICT.get(loc.status, loc.status)
lines.append(f" {icon} {loc.location}: {fps} -- {verdict}")
if s.catalog_sig_status is not None:
lines.append(
f"Catalog signature: {_CATALOG_SIG_STATUS_LINE.get(s.catalog_sig_status, s.catalog_sig_status)}"
)
lines.append("")
lines.append("What to do next:")
for step in recommend_next_steps(statuses):
lines.append(f" - {step}")
return "\n".join(lines)
+6 -10
View File
@@ -53,17 +53,13 @@ DOMAIN_PREFIX = b"bcc-release-v1|"
# the signature on every past release: verification accepts a match against # the signature on every past release: verification accepts a match against
# ANY key here. # ANY key here.
# #
# Populated by the maintainer via: # Empty until the maintainer generates the release keypair (separately from
# the catalog keypair) and pastes the public half in:
# python catalog_console.py keygen --release # python catalog_console.py keygen --release
# Rotated 2026-07 (issue #68 finding 5 / #68 CI-exposure incident): the # This is intentionally NOT pre-populated with a placeholder that looks
# original key was shared with the catalog key and had been exposed to CI, # like a real key -- release.yml's signing-smoke-test fails closed (loudly)
# so both keypairs were regenerated as separate, disjoint keys. This list # on an empty list rather than silently verifying against nothing.
# holds only the current release key -- if release.yml's signing-smoke-test RELEASE_PUBKEYS: list[bytes] = []
# ever sees this list empty, it fails closed (loudly) rather than silently
# verifying against nothing.
RELEASE_PUBKEYS: list[bytes] = [
base64.b64decode("6BnPgJEHJFyVltFoLTCNadIsehjy00iiW8IRlC1TfhA="),
]
CHUNK_SIZE = 1024 * 1024 CHUNK_SIZE = 1024 * 1024
-215
View File
@@ -1,215 +0,0 @@
"""Tests for catalog_console.py's non-Qt git plumbing and ref-resolution
seam (issue #68 rotation-completability fix).
catalog_console.py is importable here WITHOUT PySide6 -- its Qt import is
guarded (`_PYSIDE6_AVAILABLE`) precisely so `keygen`, `show-seed-b64`,
`keys`, and this git plumbing stay usable (and testable) wherever PySide6
isn't installed, including this CI test job, which never installs it. If
PySide6 genuinely isn't importable in this environment, that itself
exercises the guard path -- see test_module_imports_without_pyside6.
"""
from __future__ import annotations
import subprocess
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import catalog_console as cc
import catalog_review as review
_SEED_CATALOG = b'{"schema": 1, "version": 1, "servers": []}'
_SEED_SIG = b"\x00" * 64
def _run(*args: str, cwd: Path) -> None:
subprocess.run(["git", *args], cwd=cwd, check=True, capture_output=True)
def _init_bare_and_clone(tmp_path: Path) -> tuple[Path, Path]:
"""A bare "origin" repo with `main` and `rotation-branch` both seeded
with a catalog + (dummy) signature, plus a working clone with `origin`
already configured -- mirroring the tokened-remote clone
catalog_console.py's git plumbing is always run against."""
origin = tmp_path / "origin.git"
_run("init", "--bare", str(origin), cwd=tmp_path)
seed = tmp_path / "seed"
_run("clone", str(origin), str(seed), cwd=tmp_path)
_run("config", "user.email", "test@example.com", cwd=seed)
_run("config", "user.name", "Test", cwd=seed)
(seed / "data").mkdir()
(seed / "data" / "catalog.json").write_bytes(_SEED_CATALOG)
(seed / "data" / "catalog.json.sig").write_bytes(_SEED_SIG)
_run("add", "-A", cwd=seed)
_run("commit", "-m", "seed", cwd=seed)
_run("push", "origin", "HEAD:refs/heads/main", cwd=seed)
_run("checkout", "-b", "rotation-branch", cwd=seed)
_run("push", "origin", "HEAD:refs/heads/rotation-branch", cwd=seed)
clone = tmp_path / "work"
_run("clone", str(origin), str(clone), cwd=tmp_path)
_run("config", "user.email", "test@example.com", cwd=clone)
_run("config", "user.name", "Test", cwd=clone)
return origin, clone
# --------------------------------------------------------------------------- #
# The module must stay importable without PySide6 -- this IS the fix that
# lets `keys`/`keygen`/`show-seed-b64` (and this whole test file) run
# somewhere PySide6 isn't installed.
# --------------------------------------------------------------------------- #
def test_module_imports_without_pyside6():
assert hasattr(cc, "_PYSIDE6_AVAILABLE")
# This CI test job never installs PySide6 (see .github/workflows/ci.yml
# "Install test dependencies": pytest + cryptography only) -- so on CI,
# this assertion is itself proof the guard is doing its job. Locally,
# where a maintainer's env DOES have PySide6, it's fine either way; the
# only real assertion this test needs is "importing the module never
# raises", which happened just by getting this far.
assert cc._PYSIDE6_AVAILABLE in (True, False)
def test_cmd_gui_fails_soft_without_pyside6(monkeypatch, capsys):
if cc._PYSIDE6_AVAILABLE:
return # nothing to prove where PySide6 IS available
import argparse
args = argparse.Namespace(repo=".", ref=None)
assert cc.cmd_gui(args) == 1
assert "PySide6" in capsys.readouterr().err
# --------------------------------------------------------------------------- #
# compute_own_refs: the PURE ref-resolution seam. No git, no Qt.
# --------------------------------------------------------------------------- #
def test_compute_own_refs_defaults_to_main_only():
assert cc.compute_own_refs(None, None) == ["main"]
def test_compute_own_refs_adds_detected_branch():
assert cc.compute_own_refs(None, "chore/68-key-rotation") == [
"main",
"chore/68-key-rotation",
]
def test_compute_own_refs_explicit_ref_overrides_detected_branch():
assert cc.compute_own_refs("explicit-branch", "detected-branch") == [
"main",
"explicit-branch",
]
def test_compute_own_refs_does_not_duplicate_main():
assert cc.compute_own_refs(None, "main") == ["main"]
assert cc.compute_own_refs("main", "some-other-branch") == ["main"]
# --------------------------------------------------------------------------- #
# current_branch: git plumbing, no Qt.
# --------------------------------------------------------------------------- #
def test_current_branch_detects_checked_out_branch(tmp_path):
_origin, clone = _init_bare_and_clone(tmp_path)
_run("fetch", "origin", "rotation-branch", cwd=clone)
_run("checkout", "-B", "rotation-branch", "origin/rotation-branch", cwd=clone)
assert cc.current_branch(clone) == "rotation-branch"
def test_current_branch_none_on_detached_head(tmp_path):
_origin, clone = _init_bare_and_clone(tmp_path)
commit = cc.fetch_ref(clone, "main")
_run("checkout", commit, cwd=clone)
assert cc.current_branch(clone) is None
# --------------------------------------------------------------------------- #
# commit_and_push_signed_catalog: MUST target the given branch, never a
# hardcoded "main" -- issue #68's completability fix. This is exactly the
# bug that, before the fix, would have made ReviewWindow._on_sign push a
# PR/branch review's signature straight to main regardless of what was
# actually reviewed.
# --------------------------------------------------------------------------- #
def test_commit_and_push_signed_catalog_targets_the_given_branch_not_main(tmp_path):
_origin, clone = _init_bare_and_clone(tmp_path)
new_raw = b'{"schema": 1, "version": 2, "servers": []}'
new_sig = b"\x01" * 64
cc.commit_and_push_signed_catalog(clone, new_raw, new_sig, branch="rotation-branch")
rotation_commit = cc.fetch_ref(clone, "rotation-branch")
rotation_raw, _sha = cc.read_catalog_at_commit(clone, rotation_commit)
assert rotation_raw == new_raw
# main on the shared origin must be COMPLETELY untouched by a sign that
# was reviewed and pushed against rotation-branch.
main_commit = cc.fetch_ref(clone, "main")
main_raw, _sha = cc.read_catalog_at_commit(clone, main_commit)
assert main_raw == _SEED_CATALOG
def test_commit_and_push_signed_catalog_still_defaults_to_main(tmp_path):
"""Backward-compatible default: callers that don't pass `branch` (there
are none left in catalog_console.py itself, but the signature keeps the
default for any other caller / test fixture) still push to main."""
_origin, clone = _init_bare_and_clone(tmp_path)
new_raw = b'{"schema": 1, "version": 2, "servers": []}'
new_sig = b"\x01" * 64
cc.commit_and_push_signed_catalog(clone, new_raw, new_sig)
main_commit = cc.fetch_ref(clone, "main")
main_raw, _sha = cc.read_catalog_at_commit(clone, main_commit)
assert main_raw == new_raw
rotation_commit = cc.fetch_ref(clone, "rotation-branch")
rotation_raw, _sha = cc.read_catalog_at_commit(clone, rotation_commit)
assert rotation_raw == _SEED_CATALOG # untouched
# --------------------------------------------------------------------------- #
# catalog_sig_status_on_disk: the check behind `keys`' "does catalog.json.sig
# currently verify?" line -- this is precisely the check that would have
# caught the current chore/68-key-rotation state (bcc_core.CATALOG_PUBKEYS
# rotated, data/catalog.json.sig still signed by the retired key).
# --------------------------------------------------------------------------- #
def test_catalog_sig_status_on_disk_valid(tmp_path):
seed, pub = review.generate_keypair()
raw = b'{"schema": 1, "version": 1, "servers": []}'
sig = review.sign_catalog_bytes(raw, seed)
(tmp_path / "data").mkdir()
(tmp_path / "data" / "catalog.json").write_bytes(raw)
(tmp_path / "data" / "catalog.json.sig").write_bytes(sig)
assert cc.catalog_sig_status_on_disk(tmp_path, [pub]) == "valid"
def test_catalog_sig_status_on_disk_invalid_when_pubkey_rotated(tmp_path):
"""The exact chore/68-key-rotation scenario: signed by an OLD key, but
the committed pubkey list now only has the NEW key."""
old_seed, _old_pub = review.generate_keypair()
_new_seed, new_pub = review.generate_keypair()
raw = b'{"schema": 1, "version": 1, "servers": []}'
sig = review.sign_catalog_bytes(raw, old_seed)
(tmp_path / "data").mkdir()
(tmp_path / "data" / "catalog.json").write_bytes(raw)
(tmp_path / "data" / "catalog.json.sig").write_bytes(sig)
assert cc.catalog_sig_status_on_disk(tmp_path, [new_pub]) == "invalid"
def test_catalog_sig_status_on_disk_missing_when_no_sig_file(tmp_path):
(tmp_path / "data").mkdir()
(tmp_path / "data" / "catalog.json").write_bytes(b"{}")
assert cc.catalog_sig_status_on_disk(tmp_path, []) == "missing"
def test_catalog_sig_status_on_disk_missing_when_no_catalog_file(tmp_path):
(tmp_path / "data").mkdir()
(tmp_path / "data" / "catalog.json.sig").write_bytes(b"\x00" * 64)
assert cc.catalog_sig_status_on_disk(tmp_path, []) == "missing"
-334
View File
@@ -491,98 +491,6 @@ def test_sign_precondition_defaults_to_main_when_loaded_ref_unset():
assert decision.ok is True assert decision.ok is True
# --------------------------------------------------------------------------- #
# ReviewSession(reattest=True): key-rotation re-attestation (issue #68
# finding 5 follow-up). After rotating bcc_core.CATALOG_PUBKEYS, the
# existing data/catalog.json.sig no longer verifies under the new key even
# though catalog CONTENT is unchanged -- diff_catalogs(old, new) would be
# empty, and an empty changeset must never unlock Sign (can_sign gate 0).
# Re-attestation mode sidesteps that correctly: instead of diffing against
# the (now-untrustworthy) last-signed content, it treats every entry as
# requiring a fresh acknowledgement, exactly like a brand-new catalog.
# --------------------------------------------------------------------------- #
def test_reattest_mode_with_unchanged_content_yields_one_change_per_entry():
same = _catalog(_entry(id="a"), _entry(id="b"), _entry(id="c"))
session = r.start_review("sha1", same, same, reattest=True)
assert len(session.changes) == 3
assert {c_.entry_id for c_ in session.changes} == {"a", "b", "c"}
# Presented "as if newly added" -- old_catalog plays no role here.
assert all(c_.status == "added" for c_ in session.changes)
assert all(c_.old is None for c_ in session.changes)
def test_reattest_mode_can_sign_refuses_until_all_acknowledged_then_permits():
same = _catalog(_entry(id="a"), _entry(id="b"), _entry(id="c"))
session = r.start_review("sha1", same, same, reattest=True)
decision = r.can_sign(session, "sha1")
assert decision.ok is False
assert "acknowledged" in decision.reason.lower()
r.acknowledge_entry(session, "a")
r.acknowledge_entry(session, "b")
decision = r.can_sign(session, "sha1")
assert decision.ok is False # "c" still outstanding
assert "acknowledged" in decision.reason.lower()
r.acknowledge_entry(session, "c")
decision = r.can_sign(session, "sha1")
assert decision.ok is True
assert decision.reason is None
def test_normal_mode_with_unchanged_content_still_refuses_empty_diff():
"""The rotation path must NOT become a general bypass of the empty-diff
guard: reattest=False (the default) against identical old/new catalogs
must behave exactly as before -- can_sign refuses with 'nothing to
sign', full stop."""
same = _catalog(_entry(id="a"), _entry(id="b"))
session = r.start_review("sha1", same, same) # reattest defaults False
assert session.changes == []
decision = r.can_sign(session, "sha1")
assert decision.ok is False
assert "nothing to sign" in decision.reason.lower()
def test_reattest_mode_still_enforces_blocking_risk_check():
"""Re-attestation must not relax the blocking-risk gate: a
disallowed-command entry blocks Sign even with every entry
acknowledged."""
cat = _catalog(
_entry(id="a"),
_entry(id="evil", config={"command": "bash", "args": ["-c", "rm -rf /"]}),
)
session = r.start_review("sha1", cat, cat, reattest=True)
assert len(session.changes) == 2
r.acknowledge_entry(session, "a")
r.acknowledge_entry(session, "evil")
assert r.all_entries_acknowledged(session) is True
decision = r.can_sign(session, "sha1")
assert decision.ok is False
assert "blocking" in decision.reason.lower()
def test_reattest_mode_still_enforces_toctou_pin():
"""Re-attestation must not relax the TOCTOU blob-SHA pin: acknowledging
everything is not enough if the bytes moved underneath the review."""
same = _catalog(_entry(id="a"))
session = r.start_review("sha1", same, same, reattest=True)
r.acknowledge_entry(session, "a")
decision = r.can_sign(session, "sha1")
assert decision.ok is True # sanity: matches when blob is unchanged
decision = r.can_sign(session, "sha2-a-new-commit-landed")
assert decision.ok is False
assert "mismatch" in decision.reason.lower() or "changed" in decision.reason.lower()
def test_reattest_defaults_to_false():
"""start_review()'s reattest parameter defaults to False -- normal
(diff-based) review is the default behaviour, never silently entered."""
session = r.start_review("sha1", _catalog(), _catalog(_entry()))
assert session.reattest is False
# --------------------------------------------------------------------------- # # --------------------------------------------------------------------------- #
# find_last_signed_catalog_raw: what source=main diffs against # find_last_signed_catalog_raw: what source=main diffs against
# --------------------------------------------------------------------------- # # --------------------------------------------------------------------------- #
@@ -785,245 +693,3 @@ def test_contains_non_ascii_true():
def test_contains_non_ascii_false(): def test_contains_non_ascii_false():
assert r.contains_non_ascii("package") is False assert r.contains_non_ascii("package") is False
# --------------------------------------------------------------------------- #
# fingerprint_pubkey
# --------------------------------------------------------------------------- #
def test_fingerprint_pubkey_is_deterministic():
pub = b"\x01" * 32
assert r.fingerprint_pubkey(pub) == r.fingerprint_pubkey(pub)
def test_fingerprint_pubkey_differs_for_different_keys():
assert r.fingerprint_pubkey(b"\x01" * 32) != r.fingerprint_pubkey(b"\x02" * 32)
def test_fingerprint_pubkey_never_contains_the_key_bytes_themselves():
pub = b"\x42" * 32
fp = r.fingerprint_pubkey(pub)
assert pub.hex() not in fp.lower().replace(" ", "")
# --------------------------------------------------------------------------- #
# extract_pubkey_list_literal / extract_ci_trust_anchor_pubkey: text parsing
# for `catalog_console.py keys`, exercised here with no file I/O.
# --------------------------------------------------------------------------- #
def test_extract_pubkey_list_literal_single_key():
_seed, pub = r.generate_keypair()
import base64
text = (
"CATALOG_PUBKEYS: list[bytes] = [\n"
f' base64.b64decode("{base64.b64encode(pub).decode()}"),\n'
"]\n"
)
assert r.extract_pubkey_list_literal(text, "CATALOG_PUBKEYS") == [pub]
def test_extract_pubkey_list_literal_multiple_keys():
import base64
pubs = [r.generate_keypair()[1] for _ in range(2)]
body = ",\n".join(f' base64.b64decode("{base64.b64encode(p).decode()}")' for p in pubs)
text = f"RELEASE_PUBKEYS: list[bytes] = [\n{body},\n]\n"
assert r.extract_pubkey_list_literal(text, "RELEASE_PUBKEYS") == pubs
def test_extract_pubkey_list_literal_missing_variable_returns_empty():
assert r.extract_pubkey_list_literal("some unrelated text", "CATALOG_PUBKEYS") == []
def test_extract_pubkey_list_literal_does_not_match_a_different_variable():
import base64
_seed, pub = r.generate_keypair()
text = f'OTHER_PUBKEYS: list[bytes] = [base64.b64decode("{base64.b64encode(pub).decode()}")]\n'
assert r.extract_pubkey_list_literal(text, "CATALOG_PUBKEYS") == []
def test_extract_ci_trust_anchor_pubkey_found():
import base64
_seed, pub = r.generate_keypair()
text = f' EXPECTED_CATALOG_PUBKEY_B64: "{base64.b64encode(pub).decode()}"\n'
assert r.extract_ci_trust_anchor_pubkey(text) == pub
def test_extract_ci_trust_anchor_pubkey_missing_returns_none():
assert r.extract_ci_trust_anchor_pubkey("no anchor here") is None
# --------------------------------------------------------------------------- #
# key_status / render_key_status_report / recommend_next_steps
# --------------------------------------------------------------------------- #
def _kw(**overrides):
base = dict(
kind="catalog",
display_name="CATALOG",
purpose="Signs the catalog.",
private_key_location="on this machine",
local_exists=True,
local_pubkey=b"\x01" * 32,
locations=[("bcc_core.CATALOG_PUBKEYS", [b"\x01" * 32])],
catalog_sig_status="valid",
)
base.update(overrides)
return base
def test_key_status_reports_match_when_local_pubkey_in_committed_list():
status = r.key_status(**_kw())
assert status.locations[0].status == "match"
def test_key_status_reports_mismatch_when_local_pubkey_not_in_committed_list():
status = r.key_status(**_kw(locations=[("bcc_core.CATALOG_PUBKEYS", [b"\x02" * 32])]))
assert status.locations[0].status == "mismatch"
def test_key_status_reports_unknown_when_no_local_pubkey():
status = r.key_status(**_kw(local_pubkey=None, local_exists=False))
assert status.locations[0].status == "unknown"
assert status.local_fingerprint is None
def test_key_status_never_carries_a_local_fingerprint_when_key_absent():
status = r.key_status(**_kw(local_pubkey=None, local_exists=False))
assert status.local_exists is False
assert status.local_fingerprint is None
def test_key_status_fingerprint_matches_fingerprint_pubkey_helper():
pub = b"\x03" * 32
status = r.key_status(**_kw(local_pubkey=pub, locations=[("x", [pub])]))
assert status.local_fingerprint == r.fingerprint_pubkey(pub)
def test_key_status_checks_multiple_locations_independently():
"""A key can match one committed location and mismatch another -- this
is exactly the drift issue #68 finding 4 was about (bcc_core.py and
ci.yml silently disagreeing on the trust anchor)."""
pub = b"\x04" * 32
other = b"\x05" * 32
status = r.key_status(
**_kw(
local_pubkey=pub,
locations=[
("bcc_core.CATALOG_PUBKEYS", [pub]),
("ci.yml trust anchor", [other]),
],
)
)
assert status.locations[0].status == "match"
assert status.locations[1].status == "mismatch"
def test_recommend_next_steps_flags_never_generated_key():
status = r.key_status(**_kw(local_exists=False, local_pubkey=None, locations=[]))
steps = r.recommend_next_steps([status])
assert any("keygen" in s and "CATALOG" in s for s in steps)
def test_recommend_next_steps_release_key_uses_release_flag():
status = r.key_status(
kind="release",
display_name="RELEASE",
purpose="Signs checksums.",
private_key_location="not generated yet",
local_exists=False,
local_pubkey=None,
locations=[],
)
steps = r.recommend_next_steps([status])
assert any("keygen --release" in s for s in steps)
def test_recommend_next_steps_flags_mismatch_by_location_name():
status = r.key_status(**_kw(locations=[("bcc_core.CATALOG_PUBKEYS", [b"\x99" * 32])]))
steps = r.recommend_next_steps([status])
assert any("bcc_core.CATALOG_PUBKEYS" in s and "CATALOG" in s for s in steps)
def test_recommend_next_steps_flags_invalid_catalog_signature():
status = r.key_status(**_kw(catalog_sig_status="invalid"))
steps = r.recommend_next_steps([status])
assert any("re-signing" in s for s in steps)
def test_recommend_next_steps_flags_missing_catalog_signature():
status = r.key_status(**_kw(catalog_sig_status="missing"))
steps = r.recommend_next_steps([status])
assert any("re-signing" in s for s in steps)
def test_recommend_next_steps_all_clear_when_nothing_wrong():
status = r.key_status(**_kw())
steps = r.recommend_next_steps([status])
assert steps == ["Everything is consistent -- no action needed."]
def test_recommend_next_steps_release_key_has_no_catalog_signature_advice():
"""A mismatched RELEASE key must never trigger catalog-signing advice --
the two keys' remediation paths must not bleed into each other."""
status = r.key_status(
kind="release",
display_name="RELEASE",
purpose="Signs checksums.",
private_key_location="on this machine",
local_exists=True,
local_pubkey=b"\x06" * 32,
locations=[("scripts/sign_checksums.py RELEASE_PUBKEYS", [b"\x07" * 32])],
catalog_sig_status=None,
)
steps = r.recommend_next_steps([status])
assert not any("re-signing" in s for s in steps)
assert any("RELEASE" in s for s in steps)
def test_render_key_status_report_never_prints_private_key_material():
"""The report string must be built ONLY from public inputs. Sanity
check: no field on KeyStatus/PubkeyLocationCheck is capable of holding
private key bytes in the first place (there's no such field to leak),
and the render function only touches fields that exist -- this test
guards against a future field addition reintroducing that risk."""
status = r.key_status(**_kw())
text = r.render_key_status_report([status])
assert "CATALOG" in text
assert (
"purpose" not in text.lower() or "Signs the catalog." in text
) # sanity, not a real secret
# No 64-hex-char (or longer) run anywhere -- a raw 32-byte seed/sig
# would show up as one if it were ever accidentally interpolated in.
import re as _re
assert not _re.search(r"[0-9a-fA-F]{64,}", text)
def test_render_key_status_report_names_the_specific_key_not_generic_the_key():
status = r.key_status(**_kw())
text = r.render_key_status_report([status])
assert "CATALOG KEY" in text
assert "the key" not in text.lower()
def test_render_key_status_report_includes_catalog_signature_line_only_for_catalog():
catalog_status = r.key_status(**_kw())
release_status = r.key_status(
kind="release",
display_name="RELEASE",
purpose="Signs checksums.",
private_key_location="on this machine",
local_exists=True,
local_pubkey=b"\x08" * 32,
locations=[("scripts/sign_checksums.py RELEASE_PUBKEYS", [b"\x08" * 32])],
catalog_sig_status=None,
)
text = r.render_key_status_report([catalog_status, release_status])
assert text.count("Catalog signature:") == 1
def test_render_key_status_report_ends_with_what_to_do_next_section():
status = r.key_status(**_kw())
text = r.render_key_status_report([status])
assert "What to do next:" in text
+344
View File
@@ -2357,6 +2357,43 @@ def test_catalog_entry_to_paste_json_includes_env_when_present():
assert c.validate_catalog(malicious) != [] 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>"]}
assert c.config_has_unfilled_placeholders(cfg) is True 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(): 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"}