5 Commits

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

Finding 5: the catalog key and the release key are now separate. The release
key lives in CI and signs checksums; the catalog key stays offline and signs
what users execute. A CI compromise gets the former, not the latter.
2026-07-12 21:30:13 -04:00
the_og 26c66b7db1 Merge branch 'main' into fix/68-console-gate
CI / Tests (py3.12 / windows-latest) (pull_request) Successful in 24s
CI / Lint (ruff) (pull_request) Successful in 6s
CI / Tests (py3.10 / ubuntu-latest) (pull_request) Successful in 10s
CI / Tests (py3.12 / ubuntu-latest) (pull_request) Successful in 10s
CI / Tests (py3.13 / ubuntu-latest) (pull_request) Successful in 10s
CI / Catalog signature (pull_request) Successful in 6s
2026-07-12 21:28:27 -04:00
BCC Fix Agent 82483e693d fix(catalog-console): close the vacuous review gate; split catalog/release signing keys
CI / Lint (ruff) (pull_request) Successful in 9s
CI / Tests (py3.10 / ubuntu-latest) (pull_request) Successful in 11s
CI / Tests (py3.12 / windows-latest) (pull_request) Successful in 34s
CI / Tests (py3.12 / ubuntu-latest) (pull_request) Successful in 9s
CI / Tests (py3.13 / ubuntu-latest) (pull_request) Successful in 10s
CI / Catalog signature (pull_request) Successful in 6s
Fixes #68 findings 1 and 5.

Finding 1 -- the review gate signed without reviewing anything:
- ReviewWindow._on_sign hardcoded "main" as the TOCTOU comparison ref, so
  any PR review (where _on_load pins the PR head's blob SHA) could never
  sign; the only working path was main-vs-itself, whose empty diff made
  can_sign() vacuously True (set() <= set()). Commit b08cf21 signed 19
  entries through exactly that path with zero of them reviewed.
- can_sign() now refuses an empty changeset outright, and itself checks
  has_blocking_risk() across every changed entry rather than trusting the
  GUI to have disabled a checkbox.
- ReviewSession now carries loaded_ref (the exact ref reviewed); a new pure
  sign_precondition(session, resolve_blob_sha) resolves the TOCTOU SHA from
  that ref, never a hardcoded "main". _on_load's retry path re-diffs
  instead of re-pinning the same stale SHA, so a blob-mismatch refusal
  can't loop forever.
- source="main" now diffs against the last catalog a maintainer actually
  SIGNED (walking catalog.json's git history until a version verifies
  against the current .sig), not against itself.
- Replaced the theatre-only test_no_acknowledge_all_function_exists (only
  asserted no function was *named* acknowledge_all) with a test that also
  exercises the real gate. Added can_sign/sign_precondition coverage for
  the empty-diff, blocking-risk, and ref-resolution seams -- each verified
  to fail when its guard is removed.

Finding 5 -- the catalog key and release key were the same CI-resident key:
- scripts/sign_checksums.py gets its own RELEASE_PUBKEYS (separate from
  bcc_core.CATALOG_PUBKEYS) and a verify_checksums_against_any() helper.
- release.yml's signing-smoke-test now verifies RELEASE_SIGNING_KEY against
  RELEASE_PUBKEYS only -- it no longer imports bcc_core/CATALOG_PUBKEYS at
  all, so this workflow can never compare a CI secret against the
  catalog's root of trust.
- catalog_console.py: keygen/show-seed-b64 gain --release, with separate
  keychain/file storage per key kind. show-seed-b64 refuses to run without
  --release, so the catalog seed can't be exported to a CI secret by habit.
- README documents both keys' trust properties and the asymmetry: a CI
  compromise burns the release key, never the catalog key.

The maintainer must rotate the catalog key (it was CI-resident, so treat it
as burned for catalog use) and generate a fresh release key -- see the PR
description for the exact steps. No key is generated or committed here.
2026-07-12 21:25:34 -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
10 changed files with 1556 additions and 103 deletions
+49 -22
View File
@@ -101,9 +101,20 @@ jobs:
# signing step — which means a wrong/missing RELEASE_SIGNING_KEY secret # signing step — which means a wrong/missing RELEASE_SIGNING_KEY secret
# would only be discovered at the worst possible moment: during a real # would only be discovered at the worst possible moment: during a real
# release. This job signs a throwaway manifest with the secret and verifies # release. This job signs a throwaway manifest with the secret and verifies
# the result against the PUBLIC key already compiled into bcc_core. # the result against scripts/sign_checksums.RELEASE_PUBKEYS.
# #
# It proves the two halves of the keypair actually match, without # IMPORTANT (issue #68 finding 5): this must verify against the RELEASE
# public key, never bcc_core.CATALOG_PUBKEYS. The catalog key is the
# offline, maintainer-held root of trust for what BCC executes; it must
# NEVER be compared against a value that lives in a CI secret, because
# that comparison is itself a way to smuggle a catalog-trusted key through
# CI review ("does this repo secret match the catalog key" is a question
# this workflow must never even ask). The release key is a SEPARATE
# keypair, generated via `catalog_console.py keygen --release`, that only
# ever signs release SHA256SUMS manifests -- a CI/secret compromise burns
# this key, not the catalog key.
#
# It proves the two halves of the RELEASE keypair actually match, without
# publishing anything. Run it from the Actions tab after setting or # publishing anything. Run it from the Actions tab after setting or
# rotating the secret. # rotating the secret.
signing-smoke-test: signing-smoke-test:
@@ -120,42 +131,54 @@ jobs:
- name: Install dependencies - name: Install dependencies
run: pip install cryptography run: pip install cryptography
- name: Sign a throwaway manifest and verify against the shipped pubkey - name: Sign a throwaway manifest and verify against the RELEASE pubkey
env: env:
RELEASE_SIGNING_KEY: ${{ secrets.RELEASE_SIGNING_KEY }} RELEASE_SIGNING_KEY: ${{ secrets.RELEASE_SIGNING_KEY }}
run: | run: |
if [ -z "$RELEASE_SIGNING_KEY" ]; then if [ -z "$RELEASE_SIGNING_KEY" ]; then
echo "FAIL: RELEASE_SIGNING_KEY secret is not set." echo "FAIL: RELEASE_SIGNING_KEY secret is not set."
echo "Generate it with: python catalog_console.py show-seed-b64" echo "Generate the RELEASE key (NOT the catalog key) with:"
echo "then add it under Settings -> Actions -> Secrets." echo " python catalog_console.py keygen --release"
echo "then add its seed under Settings -> Actions -> Secrets, via:"
echo " python catalog_console.py show-seed-b64 --release"
exit 1 exit 1
fi fi
mkdir -p smoke && echo "smoke test payload" > smoke/hello.txt mkdir -p smoke && echo "smoke test payload" > smoke/hello.txt
python3 scripts/sign_checksums.py generate smoke --out smoke/SHA256SUMS python3 scripts/sign_checksums.py generate smoke --out smoke/SHA256SUMS
python3 scripts/sign_checksums.py sign --sums smoke/SHA256SUMS --out smoke/SHA256SUMS.sig python3 scripts/sign_checksums.py sign --sums smoke/SHA256SUMS --out smoke/SHA256SUMS.sig
python - <<'PY' python - <<'PY'
import base64, pathlib, sys import pathlib, sys
import bcc_core as c from scripts.sign_checksums import RELEASE_PUBKEYS, verify_checksums_against_any
from scripts.sign_checksums import verify_checksums
# The public half that ships inside the binary. If the secret is a # Deliberately does NOT import bcc_core / CATALOG_PUBKEYS at all --
# DIFFERENT key than the one users' copies trust, this fails here -- # this smoke test must never be able to compare the CI secret
# which is the entire point of the job. # against the catalog's root of trust (issue #68 finding 5). Only
pub_b64 = base64.b64encode(c.CATALOG_PUBKEYS[0]).decode() # RELEASE_PUBKEYS (scripts/sign_checksums.py) is a legitimate
# target for a CI-resident key.
if not RELEASE_PUBKEYS:
sys.exit(
"FAIL: scripts/sign_checksums.RELEASE_PUBKEYS is empty.\n"
"\n"
"Generate the release keypair with:\n"
" python catalog_console.py keygen --release\n"
"then paste the printed public key into RELEASE_PUBKEYS in\n"
"scripts/sign_checksums.py and commit that change."
)
sums = pathlib.Path("smoke/SHA256SUMS").read_text() sums = pathlib.Path("smoke/SHA256SUMS").read_text()
sig = pathlib.Path("smoke/SHA256SUMS.sig").read_bytes() sig = pathlib.Path("smoke/SHA256SUMS.sig").read_bytes()
if not verify_checksums(pub_b64, sums, sig): if not verify_checksums_against_any(RELEASE_PUBKEYS, sums, sig):
sys.exit( sys.exit(
"FAIL: the signature produced by RELEASE_SIGNING_KEY does NOT verify\n" "FAIL: the signature produced by RELEASE_SIGNING_KEY does NOT verify\n"
"against the public key in bcc_core.CATALOG_PUBKEYS.\n" "against any key in scripts/sign_checksums.RELEASE_PUBKEYS.\n"
"\n" "\n"
"The secret and the shipped public key are different keypairs. Users\n" "The secret and the shipped release public key are different keypairs.\n"
"would reject every signature this CI produces. Re-copy the seed from\n" "Downloaders would reject every signature this CI produces. Re-copy the\n"
"`catalog_console.py show-seed-b64`, or update CATALOG_PUBKEYS." "seed from `catalog_console.py show-seed-b64 --release`, or update\n"
"RELEASE_PUBKEYS with the matching public key."
) )
print("OK: RELEASE_SIGNING_KEY matches the public key shipped in bcc_core.") print("OK: RELEASE_SIGNING_KEY matches a key in RELEASE_PUBKEYS.")
PY PY
# ── Create GitHub Release with all three artifacts ────────────────────── # ── Create GitHub Release with all three artifacts ──────────────────────
@@ -206,9 +229,13 @@ jobs:
# checks. It does NOT remove Gatekeeper/SmartScreen warnings. # checks. It does NOT remove Gatekeeper/SmartScreen warnings.
# #
# The private key is a repo secret (RELEASE_SIGNING_KEY, base64 raw # The private key is a repo secret (RELEASE_SIGNING_KEY, base64 raw
# Ed25519 seed) generated via the Catalog Console (#62). If it's not # Ed25519 seed) for the RELEASE key -- a SEPARATE keypair from the
# set, we still publish the release — just without a .sig — rather # catalog key, generated via `python catalog_console.py keygen
# than fail the release outright. # --release` (issue #68 finding 5; #62). This key is intentionally
# CI-resident and signs ONLY this checksum manifest; it is never
# trusted to sign data/catalog.json. If it's not set, we still
# publish the release — just without a .sig — rather than fail the
# release outright.
- name: Check for signing key - name: Check for signing key
id: signing id: signing
run: | run: |
@@ -234,7 +261,7 @@ jobs:
- name: Warn — release will be unsigned - name: Warn — release will be unsigned
if: steps.signing.outputs.has_key != 'true' if: steps.signing.outputs.has_key != 'true'
run: | run: |
echo "::warning::RELEASE_SIGNING_KEY secret is not set — this release is being published WITHOUT a signed SHA256SUMS.sig. Add the secret (base64 raw Ed25519 seed, generated via the Catalog Console, #62) before the next tag." echo "::warning::RELEASE_SIGNING_KEY secret is not set — this release is being published WITHOUT a signed SHA256SUMS.sig. Generate the RELEASE key (python catalog_console.py keygen --release) and add its seed (python catalog_console.py show-seed-b64 --release) as this secret before the next tag."
- name: Create GitHub Release - name: Create GitHub Release
uses: softprops/action-gh-release@v2 uses: softprops/action-gh-release@v2
+45 -6
View File
@@ -36,11 +36,10 @@ are only suppressed by a paid OS-vendor certificate, which this project
doesn't have. Verifying checksums is about detecting tampering in transit or doesn't have. Verifying checksums is about detecting tampering in transit or
on a mirror, not about vouching for the software. on a mirror, not about vouching for the software.
**Release signing public key** (Ed25519, base64, raw 32 bytes): This manifest is signed with BCC's **release key**, which is a different
key from the one that signs the MCP server catalog — see
``` [Signing keys](#signing-keys) below for why, and for the public key value
<PLACEHOLDER — AJ: paste the public key from the Catalog Console (#62) here> to use with `--pubkey-b64` below.
```
### macOS / Linux ### macOS / Linux
@@ -59,7 +58,7 @@ To also verify the manifest's signature (optional, requires Python +
```bash ```bash
python3 scripts/sign_checksums.py verify \ python3 scripts/sign_checksums.py verify \
--sums SHA256SUMS --sig SHA256SUMS.sig \ --sums SHA256SUMS --sig SHA256SUMS.sig \
--pubkey-b64 "<the public key above>" --pubkey-b64 "<the release public key from Signing keys, below>"
``` ```
### Windows (PowerShell) ### Windows (PowerShell)
@@ -78,6 +77,45 @@ release is missing the `.sig` file, the checksums themselves are still
valid and safe to check against — the release workflow only skips signing, valid and safe to check against — the release workflow only skips signing,
never checksum generation. never checksum generation.
## Signing keys
BCC uses **two separate Ed25519 keypairs**, deliberately never the same
key, because they protect different things and live in different places:
| | Catalog key | Release key |
|---|---|---|
| Signs | `data/catalog.json` (the MCP server catalog every user's app trusts) | `SHA256SUMS` (the checksum manifest for release binaries) |
| Verified by | `bcc_core.CATALOG_PUBKEYS` | `scripts/sign_checksums.RELEASE_PUBKEYS` |
| Lives | Offline, passphrase-encrypted, maintainer's machine only (OS keychain or an encrypted file outside the repo — see the [Catalog Console](#files), issue #62) | A Gitea Actions repo secret, `RELEASE_SIGNING_KEY`**intentionally CI-resident** |
| 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` |
**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
the shipped catalog is only there because this key signed it. If that key
and the release-checksum key were the same (as they briefly were — see
[issue #68](../../issues/68)), then anything that can exfiltrate a Gitea
Actions secret (a malicious workflow-file PR, a compromised runner, a leaky
log) could sign a catalog every user's copy of BCC would trust, not just a
checksum manifest. Splitting them means **a CI/secret compromise burns the
release key, never the catalog key** — checksums for a future release could
be forged, which is bad, but no attacker gains the ability to make BCC run
arbitrary commands on installs that trust the catalog. That asymmetry is
the entire point of having two keys instead of one.
The catalog key is **never** meant to leave the maintainer's machine: it's
generated, stored, unlocked, and used to sign entirely inside the Catalog
Console (`catalog_console.py`), and `catalog_console.py show-seed-b64`
refuses to run without `--release` specifically so the catalog seed can't
be exported by habit or muscle memory.
**Release signing public key** (Ed25519, base64, raw 32 bytes) — this is
the RELEASE key, not the catalog key:
```
<PLACEHOLDER — AJ: paste the release public key from `catalog_console.py keygen --release` here>
```
## Run from source ## Run from source
```bash ```bash
@@ -152,6 +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`, 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={},
+203 -4
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
@@ -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 []
+256 -55
View File
@@ -12,7 +12,12 @@ 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,
or the current tip of `main`. The Console fetches the exact or the current tip of `main`. The Console fetches the exact
git blob (via a local clone's git plumbing) and PINS its git blob (via a local clone's git plumbing) and PINS its
blob SHA for the rest of this review pass. blob SHA *and the ref it came from* for the rest of this
review pass. For a PR, the diff is against `main`; for
`main`, the diff is against the last catalog a maintainer
actually SIGNED (the bytes covered by the current
data/catalog.json.sig), 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
@@ -25,20 +30,34 @@ Flow: Load -> Review -> Sign.
changed entry must be individually acknowledged (its changed entry must be individually acknowledged (its
checkbox ticked) before Sign unlocks. There is no checkbox ticked) before Sign unlocks. There is no
"acknowledge all" -- see catalog_review.py. "acknowledge all" -- see catalog_review.py.
3. Sign -- re-fetches the current blob SHA and refuses to sign unless 3. Sign -- re-resolves the current blob SHA from the SAME ref that was
it still matches the pinned SHA from step 1 (TOCTOU fix: reviewed (never a hardcoded "main") and refuses to sign
catalog_review.can_sign). On success, writes unless it still matches the pinned SHA from step 1, the diff
is non-empty, and no changed entry has an outstanding
blocking risk finding (catalog_review.sign_precondition /
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 -- so main is never red in a single commit, then pushes -- so main is never red
between a catalog merge and its signature. between a catalog merge and its signature. Signing uses the
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.
Two signing keys (issue #68 finding 5): the CATALOG key (offline,
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
CI-resident and signs ONLY the release SHA256SUMS manifest
(scripts/sign_checksums.py) -- `show-seed-b64 --release` is the only
supported way to get a seed out of this tool, and it refuses to run without
`--release` so the catalog seed can never be exported by habit. See the
README's "Signing keys" section.
""" """
from __future__ import annotations from __future__ import annotations
import argparse import argparse
import base64
import contextlib import contextlib
import getpass import getpass
import html import html
@@ -69,8 +88,28 @@ SIG_PATH = "data/catalog.json.sig"
# maintainer-only tool, so a dotfile under $HOME is an acceptable fallback # maintainer-only tool, so a dotfile under $HOME is an acceptable fallback
# when the OS keychain isn't available -- the blob stored there is always # when the OS keychain isn't available -- the blob stored there is always
# passphrase-encrypted (see catalog_review.encrypt_private_key), never raw. # passphrase-encrypted (see catalog_review.encrypt_private_key), never raw.
#
# Two SEPARATE keys are stored here, never conflated (issue #68 finding 5):
# "catalog" -- offline, Console-only. Verified by bcc_core.CATALOG_PUBKEYS.
# Roots of trust for every catalog entry BCC ships. Must
# NEVER leave this machine, never touch CI, never become an
# env var or a repo secret.
# "release" -- CI-resident. Verified by scripts.sign_checksums.
# RELEASE_PUBKEYS. Signs ONLY the release SHA256SUMS
# manifest. Its private seed is deliberately meant to be
# pasted into the RELEASE_SIGNING_KEY Gitea Actions secret
# (via `show-seed-b64 --release`) -- that is its normal,
# intended flow. A CI compromise burns this key, not the
# catalog key: that asymmetry is the whole point of having
# two keys instead of one.
KEY_STORAGE_DIR = Path.home() / ".bcc-catalog-console" KEY_STORAGE_DIR = Path.home() / ".bcc-catalog-console"
KEY_STORAGE_FILE = KEY_STORAGE_DIR / "signing_key.enc" _KEY_KINDS = ("catalog", "release")
def _key_storage_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}.enc"
HTTP_TIMEOUT = 6.0 HTTP_TIMEOUT = 6.0
@@ -97,51 +136,64 @@ def _keyring_module():
_KEYRING_SERVICE = "bcc-catalog-console" _KEYRING_SERVICE = "bcc-catalog-console"
_KEYRING_USERNAME = "signing-key"
def store_encrypted_key(blob: bytes) -> str: def _keyring_username(kind: str) -> str:
assert kind in _KEY_KINDS, f"unknown key kind {kind!r}, expected one of {_KEY_KINDS}"
return f"signing-key-{kind}"
def store_encrypted_key(blob: bytes, kind: str = "catalog") -> str:
"""Persist an already-encrypted key blob (see """Persist an already-encrypted key blob (see
catalog_review.encrypt_private_key). Prefers the OS keychain; falls back catalog_review.encrypt_private_key) under the given `kind`
to a file under KEY_STORAGE_DIR (outside the repo) with restrictive ("catalog" or "release" -- see the KEY_STORAGE_DIR comment above; the
two are stored under different keychain entries / filenames so they can
never be loaded interchangeably). Prefers the OS keychain; falls back to
a file under KEY_STORAGE_DIR (outside the repo) with restrictive
permissions. Returns a human-readable description of where it went.""" permissions. Returns a human-readable description of where it went."""
keyring = _keyring_module() keyring = _keyring_module()
if keyring is not None: if keyring is not None:
try: try:
keyring.set_password(_KEYRING_SERVICE, _KEYRING_USERNAME, blob.hex()) keyring.set_password(_KEYRING_SERVICE, _keyring_username(kind), blob.hex())
return "OS keychain (via the `keyring` package)" return "OS keychain (via the `keyring` package)"
except Exception: except Exception:
pass # fall through to the file-based path pass # fall through to the file-based path
KEY_STORAGE_DIR.mkdir(parents=True, exist_ok=True) KEY_STORAGE_DIR.mkdir(parents=True, exist_ok=True)
KEY_STORAGE_FILE.write_bytes(blob) key_file = _key_storage_file(kind)
key_file.write_bytes(blob)
with contextlib.suppress(OSError): # best-effort on platforms without POSIX perm bits with contextlib.suppress(OSError): # best-effort on platforms without POSIX perm bits
KEY_STORAGE_FILE.chmod(0o600) key_file.chmod(0o600)
return f"encrypted file at {KEY_STORAGE_FILE}" return f"encrypted file at {key_file}"
def load_encrypted_key() -> bytes: def load_encrypted_key(kind: str = "catalog") -> bytes:
"""Load the encrypted key blob from wherever store_encrypted_key() put """Load the encrypted key blob of the given `kind` from wherever
it. Raises FileNotFoundError if no key has been generated yet.""" store_encrypted_key() put it. Raises FileNotFoundError if no key of that
kind has been generated yet."""
keyring = _keyring_module() keyring = _keyring_module()
if keyring is not None: if keyring is not None:
try: try:
hex_blob = keyring.get_password(_KEYRING_SERVICE, _KEYRING_USERNAME) hex_blob = keyring.get_password(_KEYRING_SERVICE, _keyring_username(kind))
if hex_blob: if hex_blob:
return bytes.fromhex(hex_blob) return bytes.fromhex(hex_blob)
except Exception: except Exception:
pass pass
if not KEY_STORAGE_FILE.exists(): key_file = _key_storage_file(kind)
if not key_file.exists():
flag = " --release" if kind == "release" else ""
raise FileNotFoundError( raise FileNotFoundError(
f"No signing key found (checked the OS keychain and {KEY_STORAGE_FILE}). " f"No {kind} signing key found (checked the OS keychain and {key_file}). "
"Run `python catalog_console.py keygen` first." f"Run `python catalog_console.py keygen{flag}` first."
) )
return KEY_STORAGE_FILE.read_bytes() return key_file.read_bytes()
def unlock_signing_key(passphrase: str) -> bytes: def unlock_signing_key(passphrase: str, kind: str = "catalog") -> bytes:
"""Load + decrypt the signing key seed. Raises ValueError on a wrong """Load + decrypt the signing key seed of the given `kind`. Raises
passphrase, FileNotFoundError if no key exists yet.""" ValueError on a wrong passphrase, FileNotFoundError if no key of that
blob = load_encrypted_key() kind exists yet. Defaults to "catalog" because that's the key
ReviewWindow._on_sign uses -- the GUI never touches the release key."""
blob = load_encrypted_key(kind)
return review.decrypt_private_key(blob, passphrase) return review.decrypt_private_key(blob, passphrase)
@@ -188,6 +240,49 @@ def read_catalog_at_commit(repo_dir: Path, commit: str) -> tuple[bytes, str]:
return blob_bytes(repo_dir, sha), sha return blob_bytes(repo_dir, sha), sha
def catalog_blob_history(repo_dir: Path, ref: str, limit: int = 200) -> list[str]:
"""Blob SHAs of CATALOG_PATH at each commit that touched it, walking
back from `ref`, most-recent-first. Used by last_signed_catalog_raw() to
find "the version of the catalog the current signature actually covers"
without assuming it's the tip commit."""
log_out = _git(repo_dir, "log", f"--max-count={limit}", "--format=%H", ref, "--", CATALOG_PATH)
commits = [line for line in log_out.splitlines() if line]
shas: list[str] = []
for commit in commits:
try:
shas.append(blob_sha_at(repo_dir, commit, CATALOG_PATH))
except GitError:
continue
return shas
def last_signed_catalog_raw(repo_dir: Path, commit: str) -> bytes | None:
"""The raw data/catalog.json bytes that verify against
data/catalog.json.sig as of `commit` -- i.e. "the last catalog a
maintainer actually signed", found by walking the catalog's git history
on that ref until a version verifies against the *current* signature.
This is what source="main (current tip)" diffs against (issue #68
finding 1): if main's tip catalog.json already matches its .sig, this
returns that same content and the diff is correctly empty (nothing new
to sign). If someone merged a catalog change to main without running it
through the Console -- the exact bypass that produced commit b08cf21 --
the signature still covers the OLDER content, so this returns that older
version and the diff surfaces exactly what was never actually reviewed.
Returns None if there's no signature yet, or none of the recent history
verifies against it (caller should treat this as "diff against nothing
ever signed", i.e. every current entry shows as newly added).
"""
try:
sig_sha = blob_sha_at(repo_dir, commit, SIG_PATH)
except GitError:
return None
sig_bytes = blob_bytes(repo_dir, sig_sha)
candidates = [blob_bytes(repo_dir, sha) for sha in catalog_blob_history(repo_dir, commit)]
return review.find_last_signed_catalog_raw(candidates, 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:
@@ -644,28 +739,50 @@ class ReviewWindow(QMainWindow):
row = self.source_list.currentRow() row = self.source_list.currentRow()
try: try:
if row <= 0: if row <= 0:
commit = fetch_ref(self.repo_dir, "main") # source = main: diff against the last catalog a maintainer
old_commit = None # main vs itself has no "old" -- nothing to diff without a base # actually SIGNED (the bytes covered by the current
# data/catalog.json.sig), never against itself. Diffing
# main-vs-main is what made an empty diff -> instantly
# "signable" in the first place (issue #68 finding 1) --
# this is the only path that reaches sign_catalog_bytes(),
# so if it can't be trusted nothing can.
loaded_ref = "main"
commit = fetch_ref(self.repo_dir, loaded_ref)
new_raw, new_blob_sha = read_catalog_at_commit(self.repo_dir, commit)
new_catalog = core.load_catalog(new_raw)
old_raw = last_signed_catalog_raw(self.repo_dir, commit)
old_catalog = (
core.load_catalog(old_raw)
if old_raw is not None
else {
"schema": 1,
"version": 0,
"servers": [],
}
)
else: else:
pr = self._prs[row - 1] pr = self._prs[row - 1]
commit = fetch_ref(self.repo_dir, pr.head_ref) loaded_ref = pr.head_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")
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 old_commit:
old_raw, _old_sha = read_catalog_at_commit(self.repo_dir, old_commit) old_raw, _old_sha = read_catalog_at_commit(self.repo_dir, old_commit)
old_catalog = core.load_catalog(old_raw) old_catalog = core.load_catalog(old_raw)
else:
old_catalog = new_catalog
except (GitError, ValueError) as e: except (GitError, ValueError) as e:
QMessageBox.critical(self, "Load failed", html.escape(str(e))) QMessageBox.critical(self, "Load failed", html.escape(str(e)))
return return
self._new_raw = new_raw self._new_raw = new_raw
self.session = review.start_review(new_blob_sha, old_catalog, new_catalog) # `loaded_ref` is pinned into the session (not just this method's
# local variable) so _on_sign can re-resolve the TOCTOU blob SHA
# from the SAME ref that was reviewed, instead of a hardcoded
# "main" -- see review.sign_precondition and issue #68 finding 1.
self.session = review.start_review(
new_blob_sha, old_catalog, new_catalog, loaded_ref=loaded_ref
)
self._render_cards() self._render_cards()
def _render_cards(self): def _render_cards(self):
@@ -706,20 +823,43 @@ class ReviewWindow(QMainWindow):
def _on_sign(self): def _on_sign(self):
assert self.session is not None assert self.session is not None
def _resolve_blob_sha(ref: str) -> str:
# Re-fetch fresh, immediately before signing, from the SAME ref
# that was reviewed (session.loaded_ref) -- NEVER hardcode
# "main" here. Hardcoding "main" is the bug that made the PR
# review path unable to sign at all: _on_load() pins the PR
# head's blob SHA, so comparing against main's SHA differs by
# definition for any PR that actually changes the catalog, and
# this refused every PR review permanently (see issue #68
# finding 1 and review.sign_precondition's docstring).
return blob_sha_at(self.repo_dir, fetch_ref(self.repo_dir, ref), CATALOG_PATH)
try: try:
current_sha = blob_sha_at(self.repo_dir, fetch_ref(self.repo_dir, "main"), CATALOG_PATH) decision = review.sign_precondition(self.session, _resolve_blob_sha)
except GitError as e: except GitError as e:
QMessageBox.critical(self, "Sign failed", html.escape(str(e))) QMessageBox.critical(self, "Sign failed", html.escape(str(e)))
return return
decision = review.can_sign(self.session, current_sha)
if not decision.ok: if not decision.ok:
QMessageBox.warning(self, "Cannot sign", html.escape(decision.reason or "")) QMessageBox.warning(self, "Cannot sign", html.escape(decision.reason or ""))
if decision.reason and "changed" in decision.reason.lower(): # Only the TOCTOU blob-SHA-mismatch reason should trigger a
self._on_load() # force a re-review against the new bytes # reload -- "mismatch" appears ONLY in that reason (deliberately
# checked instead of the broader "changed", which also matches
# "Not every CHANGED entry has been acknowledged yet" and would
# wrongly force a reload -- and with it a fresh review.py
# ReviewSession -- every time a reviewer pauses partway through
# ticking checkboxes).
if decision.reason and "mismatch" in decision.reason.lower():
# Force a genuine re-review against the new bytes -- this
# must call _on_load() (which re-fetches and re-diffs), NOT
# re-invoke the same stale resolver, or this becomes the
# infinite loop described in issue #68 finding 1: re-pinning
# the same wrong SHA forever instead of ever converging.
self._on_load()
return return
dialog = PassphraseDialog("Enter signing key passphrase:", self) dialog = PassphraseDialog("Enter CATALOG signing key passphrase:", self)
if dialog.exec() != QDialog.DialogCode.Accepted: if dialog.exec() != QDialog.DialogCode.Accepted:
return return
try: try:
@@ -744,9 +884,17 @@ class ReviewWindow(QMainWindow):
# --------------------------------------------------------------------------- # # --------------------------------------------------------------------------- #
def cmd_keygen(_args: argparse.Namespace) -> int: def cmd_keygen(args: argparse.Namespace) -> int:
# Two SEPARATE keypairs, never conflated (issue #68 finding 5): the
# catalog key is the offline root of trust for what BCC executes and
# must never touch CI; the release key is CI-resident and signs ONLY
# the release SHA256SUMS manifest. Which one this run generates is
# explicit via --release, and the printed instructions differ sharply
# so it's obvious which key is safe to paste into a CI secret (release)
# and which one never is (catalog).
kind = "release" if getattr(args, "release", False) else "catalog"
seed, pubkey = review.generate_keypair() seed, pubkey = review.generate_keypair()
passphrase = getpass.getpass("Choose a passphrase to encrypt the new signing key: ") passphrase = getpass.getpass(f"Choose a passphrase to encrypt the new {kind} signing key: ")
confirm = getpass.getpass("Confirm passphrase: ") confirm = getpass.getpass("Confirm passphrase: ")
if passphrase != confirm: if passphrase != confirm:
print("error: passphrases did not match", file=sys.stderr) print("error: passphrases did not match", file=sys.stderr)
@@ -756,31 +904,67 @@ def cmd_keygen(_args: argparse.Namespace) -> int:
return 1 return 1
blob = review.encrypt_private_key(seed, passphrase) blob = review.encrypt_private_key(seed, passphrase)
where = store_encrypted_key(blob) where = store_encrypted_key(blob, kind=kind)
pubkey_b64 = __import__("base64").b64encode(pubkey).decode("ascii") pubkey_b64 = base64.b64encode(pubkey).decode("ascii")
print(f"Private key encrypted and stored in: {where}") print(f"{kind.capitalize()} private key encrypted and stored in: {where}")
print() print()
print("Public key (base64, paste into bcc_core.CATALOG_PUBKEYS):") if kind == "catalog":
print(
"This is the CATALOG key. It is the root of trust for every catalog "
"entry BCC ships -- it must stay offline and Console-only. NEVER paste "
"it, its seed, or `show-seed-b64` output into CI, an env var, or a repo "
"secret. (If the key currently in bcc_core.CATALOG_PUBKEYS has ever been "
"pasted into a CI secret, treat it as burned for catalog use -- generate "
"a fresh one with this command and rotate.)"
)
print()
print(
"Public key (base64) -- hand this to whoever maintains bcc_core.py so "
"they can add it to CATALOG_PUBKEYS (this tool does not edit that file):"
)
print(f" {pubkey_b64}")
else:
print(
"This is the RELEASE key. It signs ONLY the release SHA256SUMS "
"manifest in CI -- it is intentionally CI-resident and is NOT trusted "
"to sign the catalog (bcc_core.CATALOG_PUBKEYS does not and must not "
"contain it)."
)
print()
print("1. Public key (base64) -- paste into scripts/sign_checksums.py RELEASE_PUBKEYS:")
print(f" {pubkey_b64}") print(f" {pubkey_b64}")
print() print()
print( print(
"Also add it as the Gitea repo secret RELEASE_SIGNING_KEY (base64 of the " "2. Private key -- add it as the Gitea repo secret RELEASE_SIGNING_KEY "
"32-byte private seed) used by release.yml -- get that value with:" "(base64 of the 32-byte private seed). Get that value with:"
) )
print(" python catalog_console.py show-seed-b64 # careful: prints the raw key") print(" python catalog_console.py show-seed-b64 --release # prints the raw key")
return 0 return 0
def cmd_show_seed_b64(_args: argparse.Namespace) -> int: def cmd_show_seed_b64(args: argparse.Namespace) -> int:
passphrase = getpass.getpass("Signing key passphrase: ") # Deliberately requires --release: this command's whole purpose is to
# produce a value that gets pasted into a CI secret, and the catalog key
# must NEVER be pasted into CI (issue #68 finding 5 -- that is exactly
# how the catalog key ended up burned in the first place). Refusing to
# run without --release makes "export the catalog seed for CI" a
# structurally different, more deliberate action than a typo away.
if not getattr(args, "release", False):
print(
"error: show-seed-b64 only ever exports the RELEASE key -- it is the "
"only key allowed to leave this machine, for the RELEASE_SIGNING_KEY CI "
"secret. Re-run as `show-seed-b64 --release`. The catalog key must never "
"be exported this way; see issue #68 finding 5.",
file=sys.stderr,
)
return 1
passphrase = getpass.getpass("Release signing key passphrase: ")
try: try:
seed = unlock_signing_key(passphrase) 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
import base64
print(base64.b64encode(seed).decode("ascii")) print(base64.b64encode(seed).decode("ascii"))
return 0 return 0
@@ -810,11 +994,28 @@ def build_parser() -> argparse.ArgumentParser:
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.set_defaults(func=cmd_gui) p_gui.set_defaults(func=cmd_gui)
p_keygen = sub.add_parser("keygen", help="generate a new Ed25519 signing keypair") p_keygen = sub.add_parser(
"keygen", help="generate a new Ed25519 signing keypair (catalog key by default)"
)
p_keygen.add_argument(
"--release",
action="store_true",
help=(
"generate the RELEASE key (CI-resident, signs SHA256SUMS only) instead "
"of the CATALOG key (offline, Console-only, signs data/catalog.json -- "
"see issue #68 finding 5)"
),
)
p_keygen.set_defaults(func=cmd_keygen) p_keygen.set_defaults(func=cmd_keygen)
p_seed = sub.add_parser( p_seed = sub.add_parser(
"show-seed-b64", help="print the base64 private seed (for the RELEASE_SIGNING_KEY secret)" "show-seed-b64",
help="print a base64 private seed for a CI secret -- RELEASE key only",
)
p_seed.add_argument(
"--release",
action="store_true",
help="required: only the release key may ever be exported this way",
) )
p_seed.set_defaults(func=cmd_show_seed_b64) p_seed.set_defaults(func=cmd_show_seed_b64)
+97 -6
View File
@@ -24,6 +24,7 @@ from urllib.parse import urlsplit
from bcc_core import _CATALOG_SIG_DOMAIN as CATALOG_SIG_DOMAIN from bcc_core import _CATALOG_SIG_DOMAIN as CATALOG_SIG_DOMAIN
from bcc_core import CATALOG_ALLOWED_COMMANDS from bcc_core import CATALOG_ALLOWED_COMMANDS
from bcc_core import verify_catalog_signature as _verify_catalog_signature
# --------------------------------------------------------------------------- # # --------------------------------------------------------------------------- #
# Semantic diff # Semantic diff
@@ -432,11 +433,21 @@ def has_blocking_risk(change: EntryChange) -> bool:
class ReviewSession: class ReviewSession:
"""State for one review pass. `pinned_blob_sha` is the git blob SHA of """State for one review pass. `pinned_blob_sha` is the git blob SHA of
data/catalog.json as it existed the moment review began -- see data/catalog.json as it existed the moment review began -- see
can_sign().""" can_sign()/sign_precondition().
`loaded_ref` is the exact ref this review was loaded from ("main", or a
PR's `refs/pull/<n>/head`) -- see issue #68 finding 1. It exists so the
Sign path can re-resolve the TOCTOU blob SHA from *the ref that was
actually reviewed*, instead of a hardcoded "main" that silently diverges
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
main-vs-itself path instead).
"""
pinned_blob_sha: str pinned_blob_sha: str
old_catalog: dict old_catalog: dict
new_catalog: dict new_catalog: dict
loaded_ref: str = "main"
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)
@@ -445,12 +456,39 @@ class ReviewSession:
self.changes = diff_catalogs(self.old_catalog, self.new_catalog) self.changes = diff_catalogs(self.old_catalog, self.new_catalog)
def start_review(pinned_blob_sha: str, old_catalog: dict, new_catalog: dict) -> ReviewSession: def start_review(
pinned_blob_sha: str,
old_catalog: dict,
new_catalog: dict,
loaded_ref: str = "main",
) -> ReviewSession:
return ReviewSession( return ReviewSession(
pinned_blob_sha=pinned_blob_sha, old_catalog=old_catalog, new_catalog=new_catalog pinned_blob_sha=pinned_blob_sha,
old_catalog=old_catalog,
new_catalog=new_catalog,
loaded_ref=loaded_ref,
) )
def find_last_signed_catalog_raw(
candidates: list[bytes], sig: bytes, pubkeys: list[bytes]
) -> bytes | None:
"""Given `candidates` (candidate raw catalog.json byte-strings -- e.g.
successive historical versions from git log, most-recent-first),
return the first one whose signature verifies against `sig`/`pubkeys`,
or None if none do.
This is how source="main" review diffs against "the last catalog a
maintainer actually signed" instead of against itself (issue #68
finding 1): `catalog_console.last_signed_catalog_raw` walks
data/catalog.json's git history on main and hands the candidates here.
"""
for raw in candidates:
if _verify_catalog_signature(raw, sig, pubkeys):
return raw
return None
def acknowledge_entry(session: ReviewSession, entry_id: str) -> None: def acknowledge_entry(session: ReviewSession, entry_id: str) -> None:
ids = {c.entry_id for c in session.changes} ids = {c.entry_id for c in session.changes}
if entry_id not in ids: if entry_id not in ids:
@@ -483,22 +521,53 @@ class SignDecision:
def can_sign(session: ReviewSession, current_blob_sha: str) -> SignDecision: def can_sign(session: ReviewSession, current_blob_sha: str) -> SignDecision:
"""Whether the Sign button may fire right now. """Whether the Sign button may fire right now.
Two independent gates, both required: Four independent gates, all required, checked in this order:
1. TOCTOU: `current_blob_sha` (fetched fresh, immediately before signing)
0. The diff must be non-empty. An empty diff historically meant "Sign
unlocks instantly" (`set() <= set()` is vacuously True), which is
exactly backwards: a vacuously-satisfied gate is worse than no gate
at all, because it *manufactures confidence* -- the signature looks
identical to one produced by a real review. "Nothing changed" must
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
ever reviewed.)
1. TOCTOU: `current_blob_sha` (fetched fresh, immediately before signing,
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
remote changed since -- a new commit pushed to the same PR, a remote changed since -- a new commit pushed to the same PR, a
force-push, another PR merged in between -- signing is refused and a force-push, another PR merged in between -- signing is refused and a
re-review is forced. This is what makes "signing is the approval act" re-review is forced. This is what makes "signing is the approval act"
true rather than aspirational: the signature is bound to the exact true rather than aspirational: the signature is bound to the exact
reviewed bytes, not to "whatever the file happens to be now". reviewed bytes, not to "whatever the file happens to be now".
2. Every changed entry in the diff must be individually acknowledged. 2. No blocking risk finding may be outstanding on ANY changed entry, full
stop -- checked here, not just in the GUI. The GUI additionally
disables the acknowledge checkbox for a blocking entry, but that is a
UI nicety, not the enforcement point: if this pure gate didn't also
check it, a blocking risk would only be stopped by the GUI happening
to have wired the checkbox correctly, and nothing would catch a
regression in that wiring. The GUI must not be the only thing
standing between a blocking risk and a signature.
3. Every changed entry in the diff must be individually acknowledged.
""" """
if not session.changes:
return SignDecision(
False,
"Nothing to sign: this review's diff is empty. If you expected "
"changes here, you may be diffing the wrong source/ref.",
)
if current_blob_sha != session.pinned_blob_sha: if current_blob_sha != session.pinned_blob_sha:
return SignDecision( return SignDecision(
False, False,
"The reviewed bytes changed since this review began (blob SHA " "The reviewed bytes changed since this review began (blob SHA "
"mismatch) -- re-review required before signing.", "mismatch) -- re-review required before signing.",
) )
blocking_ids = sorted({c.entry_id for c in session.changes if has_blocking_risk(c)})
if blocking_ids:
return SignDecision(
False,
"Blocking risk finding(s) outstanding on: "
f"{', '.join(blocking_ids)} -- fix the underlying change, do not sign around it.",
)
if not all_entries_acknowledged(session): if not all_entries_acknowledged(session):
pending = sorted({c.entry_id for c in session.changes} - session.acknowledged) pending = sorted({c.entry_id for c in session.changes} - session.acknowledged)
return SignDecision( return SignDecision(
@@ -507,6 +576,28 @@ def can_sign(session: ReviewSession, current_blob_sha: str) -> SignDecision:
return SignDecision(True, None) return SignDecision(True, None)
def sign_precondition(
session: ReviewSession, resolve_blob_sha: Callable[[str], str]
) -> SignDecision:
"""The real Sign-button gate: resolves the current TOCTOU blob SHA from
*the ref this session was actually loaded from* (`session.loaded_ref`),
never a hardcoded "main", then delegates to can_sign().
`resolve_blob_sha` is injected so this stays testable without git/Qt --
catalog_console.ReviewWindow._on_sign passes a real resolver
(fetch_ref + blob_sha_at against self.repo_dir); tests pass a fake
dict-backed lookup. This is the fix for issue #68 finding 1's first bug:
`_on_sign` used to hardcode `fetch_ref(self.repo_dir, "main")` as the
comparison ref, so for any PR review (where `loaded_ref` is the PR's
head, not main) the SHAs differed by definition and Sign could never
fire -- and the retry path re-called the same hardcoded resolver, so it
re-pinned the same wrong value and looped forever instead of forcing a
genuine re-review.
"""
current_blob_sha = resolve_blob_sha(session.loaded_ref)
return can_sign(session, current_blob_sha)
def catalog_signing_message(raw_bytes: bytes) -> bytes: def catalog_signing_message(raw_bytes: bytes) -> bytes:
"""The exact bytes that get signed: bcc_core's domain-separation prefix """The exact bytes that get signed: bcc_core's domain-separation prefix
(imported, never retyped) + the raw catalog bytes. Using this function (imported, never retyped) + the raw catalog bytes. Using this function
+32
View File
@@ -42,6 +42,25 @@ from pathlib import Path
# message signed by the same key. # message signed by the same key.
DOMAIN_PREFIX = b"bcc-release-v1|" DOMAIN_PREFIX = b"bcc-release-v1|"
# Public half of the RELEASE signing key(s) -- a SEPARATE keypair from
# bcc_core.CATALOG_PUBKEYS (issue #68 finding 5). The catalog key is the
# offline, Console-only root of trust for what BCC executes; this key is
# CI-resident and signs ONLY the release SHA256SUMS manifest, never the
# catalog. Keeping them apart means a CI/repo-secret compromise burns the
# release key -- annoying, but it never lets an attacker sign a catalog a
# user's binary would trust. A LIST (not a single key), mirroring
# CATALOG_PUBKEYS, so the release key can be rotated without invalidating
# the signature on every past release: verification accepts a match against
# ANY key here.
#
# 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
# This is intentionally NOT pre-populated with a placeholder that looks
# like a real key -- release.yml's signing-smoke-test fails closed (loudly)
# on an empty list rather than silently verifying against nothing.
RELEASE_PUBKEYS: list[bytes] = []
CHUNK_SIZE = 1024 * 1024 CHUNK_SIZE = 1024 * 1024
@@ -135,6 +154,19 @@ def public_key_b64_from_seed(seed_b64: str) -> str:
return base64.b64encode(raw).decode("ascii") return base64.b64encode(raw).decode("ascii")
def verify_checksums_against_any(pubkeys: list[bytes], sums_text: str, signature: bytes) -> bool:
"""Verify `signature` against ANY key in `pubkeys` (each a raw 32-byte
Ed25519 public key). Mirrors bcc_core.verify_catalog_signature's
rotation-friendly "any currently-trusted key" semantics, applied to
RELEASE_PUBKEYS instead of the catalog's key list. Returns False (never
raises) for an empty `pubkeys` list -- fails closed rather than
vacuously verifying against nothing."""
return any(
verify_checksums(base64.b64encode(pk).decode("ascii"), sums_text, signature)
for pk in pubkeys
)
# --------------------------------------------------------------------------- # # --------------------------------------------------------------------------- #
# CLI # CLI
# --------------------------------------------------------------------------- # # --------------------------------------------------------------------------- #
+143 -3
View File
@@ -334,13 +334,29 @@ def test_acknowledge_gating_requires_every_entry():
assert r.all_entries_acknowledged(session) is True assert r.all_entries_acknowledged(session) is True
def test_no_acknowledge_all_function_exists(): def test_no_acknowledge_all_shortcut_and_gate_is_real():
"""Deliberate: there must be no shortcut to acknowledge every entry at """Two things, both load-bearing (issue #68: the original version of
once. See the comment in catalog_review.py above SignDecision.""" this test asserted ONLY the first half, and passed the entire time the
gate below it was vacuously satisfiable -- 'no function named
acknowledge_all' is worthless if signing doesn't actually require
acknowledgement in practice).
1. No bulk-acknowledge shortcut exists (see the comment in
catalog_review.py above SignDecision -- deliberate friction).
2. The gate that friction protects is actually enforced: with entries
still unacknowledged, can_sign() must refuse, not just "some GUI
checkbox happens to be unticked".
"""
names = [n for n in dir(r) if "acknowledge" in n.lower()] names = [n for n in dir(r) if "acknowledge" in n.lower()]
assert "acknowledge_all" not in names assert "acknowledge_all" not in names
assert "acknowledge_all_entries" not in names assert "acknowledge_all_entries" not in names
session = r.start_review("sha1", _catalog(), _catalog(_entry(id="a"), _entry(id="b")))
r.acknowledge_entry(session, "a") # only one of two -- not a bulk call
decision = r.can_sign(session, "sha1")
assert decision.ok is False
assert "acknowledged" in decision.reason.lower()
# --------------------------------------------------------------------------- # # --------------------------------------------------------------------------- #
# can_sign: TOCTOU blob pinning + acknowledge gating combined # can_sign: TOCTOU blob pinning + acknowledge gating combined
@@ -377,6 +393,130 @@ def test_can_sign_blob_mismatch_takes_priority_message():
assert "blob" in decision.reason.lower() or "changed" in decision.reason.lower() assert "blob" in decision.reason.lower() or "changed" in decision.reason.lower()
def test_can_sign_false_on_empty_changeset():
"""The exact bug behind issue #68 finding 1: 'main' loaded against
itself diffs to [], and an empty changeset used to leave can_sign()
with nothing to refuse on (set() <= set() is vacuously True). Commit
b08cf21 signed 19 entries through precisely this path -- zero of them
were ever reviewed. An empty diff must mean 'nothing to sign', never
'sign unlocked'."""
same_catalog = _catalog(_entry())
session = r.start_review("sha1", same_catalog, same_catalog)
assert session.changes == [] # diff_catalogs(x, x) -> []
assert r.all_entries_acknowledged(session) is True # vacuously -- this is the trap
decision = r.can_sign(session, "sha1") # blob matches, "everything" acknowledged
assert decision.ok is False
assert "nothing to sign" in decision.reason.lower()
def test_can_sign_false_with_outstanding_blocking_risk_even_if_acknowledged():
"""can_sign() must itself refuse a blocking risk finding -- today a
blocking finding only disables the GUI checkbox, so the pure gate must
not simply trust that the caller never acknowledged a blocking entry.
Acknowledge it directly here (bypassing any GUI checkbox-disable logic
entirely) to prove the gate catches it independently of the GUI."""
session = r.start_review(
"sha1",
_catalog(),
_catalog(_entry(config={"command": "bash", "args": ["-c", "evil"]})),
)
r.acknowledge_entry(session, "filesystem")
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()
# --------------------------------------------------------------------------- #
# sign_precondition: the ref-resolution seam that used to hardcode "main"
# --------------------------------------------------------------------------- #
def test_sign_precondition_resolves_against_loaded_ref_not_hardcoded_main():
"""The regression test for issue #68 finding 1's first bug:
ReviewWindow._on_sign used to hardcode fetch_ref(repo, "main") as the
TOCTOU comparison ref. For a PR review, _on_load pins the PR HEAD's
blob SHA, so comparing against main's SHA differs by definition and
Sign could never fire on the PR path.
The fake resolver below returns a DIFFERENT (deliberately wrong) SHA for
"main" than for the PR ref that was actually loaded. If
sign_precondition ever resolves against "main" instead of
session.loaded_ref, this test fails -- both via the recorded `calls`
list and via decision.ok flipping to False.
"""
pr_ref = "refs/pull/42/head"
session = r.start_review("pr-blob-sha", _catalog(), _catalog(_entry()), loaded_ref=pr_ref)
r.acknowledge_entry(session, "filesystem")
calls: list[str] = []
def fake_resolver(ref: str) -> str:
calls.append(ref)
return {"main": "main-blob-sha-WRONG", pr_ref: "pr-blob-sha"}[ref]
decision = r.sign_precondition(session, fake_resolver)
assert calls == [pr_ref] # never asked the resolver for "main"
assert decision.ok is True
assert decision.reason is None
def test_sign_precondition_refuses_when_loaded_ref_blob_moved():
"""Same seam, the negative case: if the loaded ref's blob SHA has moved
since review began (a new commit landed on the reviewed PR/branch), the
resolver reflects that and sign_precondition must refuse -- proving this
isn't just a hardcoded pass-through."""
pr_ref = "refs/pull/42/head"
session = r.start_review("pr-blob-sha", _catalog(), _catalog(_entry()), loaded_ref=pr_ref)
r.acknowledge_entry(session, "filesystem")
def fake_resolver(_ref: str) -> str:
return "pr-blob-sha-AFTER-A-NEW-PUSH"
decision = r.sign_precondition(session, fake_resolver)
assert decision.ok is False
assert "mismatch" in decision.reason.lower() or "changed" in decision.reason.lower()
def test_sign_precondition_defaults_to_main_when_loaded_ref_unset():
"""start_review()'s loaded_ref defaults to 'main' for source=main
reviews (and backward-compat with callers that don't pass it)."""
session = r.start_review("sha1", _catalog(), _catalog(_entry()))
assert session.loaded_ref == "main"
r.acknowledge_entry(session, "filesystem")
def fake_resolver(ref: str) -> str:
assert ref == "main"
return "sha1"
decision = r.sign_precondition(session, fake_resolver)
assert decision.ok is True
# --------------------------------------------------------------------------- #
# find_last_signed_catalog_raw: what source=main diffs against
# --------------------------------------------------------------------------- #
def test_find_last_signed_catalog_raw_returns_matching_candidate():
"""Simulates walking catalog.json's git history: the CURRENT signature
covers an OLDER version of the bytes (a later commit changed
catalog.json without re-signing -- the exact bypass that produced
commit b08cf21). The first candidate that verifies against that
signature is 'the last catalog a maintainer actually signed'."""
seed, pubkey = r.generate_keypair()
old_raw = b'{"schema":1,"version":1,"servers":[]}'
new_raw = b'{"schema":1,"version":2,"servers":[]}'
sig = r.sign_catalog_bytes(old_raw, seed) # signature covers the OLD bytes
found = r.find_last_signed_catalog_raw([new_raw, old_raw], sig, [pubkey])
assert found == old_raw
def test_find_last_signed_catalog_raw_none_when_nothing_verifies():
seed, _pubkey = r.generate_keypair()
_other_seed, other_pubkey = r.generate_keypair()
raw = b'{"schema":1,"version":1,"servers":[]}'
sig = r.sign_catalog_bytes(raw, seed)
# Check against a pubkey list that does NOT include the signer's key.
assert r.find_last_signed_catalog_raw([raw], sig, [other_pubkey]) is None
# --------------------------------------------------------------------------- # # --------------------------------------------------------------------------- #
# catalog_signing_message: domain separation must match bcc_core exactly # catalog_signing_message: domain separation must match bcc_core exactly
# --------------------------------------------------------------------------- # # --------------------------------------------------------------------------- #
+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"}