the_og/better-claude-config
1
0
Fork 0
Code Issues 8 Pull Requests 3 Actions Packages Projects Releases 4 Wiki Activity

feat: Catalog Console -- maintainer-only review + signing tool (#62) #66

Merged
the_og merged 2 commits from feat/62-catalog-console into main 2026-07-12 18:04:54 -04:00
Conversation 0 Commits 2 Files Changed 4
+2187
4 changed files with 2187 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
catalog_console.py
+839
View File
@@ -0,0 +1,839 @@
"""
catalog_console.py -- Catalog Console: maintainer-only review + signing tool
for data/catalog.json (issue #62).
MAINTAINER-ONLY. Run from a source checkout. NEVER shipped to users and
NEVER included in the release bundle -- see bcc.spec (Analysis only ever
starts from bcc.py) and tests/test_packaging.py, which asserts this file
and catalog_review.py are absent from the packaged bundle.
Flow: Load -> Review -> Sign.
1. Load -- pick a source: an open Gitea PR touching data/catalog.json,
or the current tip of `main`. The Console fetches the exact
git blob (via a local clone's git plumbing) and PINS its
blob SHA for the rest of this review pass.
2. Review -- a semantic diff (catalog_review.diff_catalogs), one card per
changed entry, with risk annotations
(catalog_review.entry_risk_findings). A registry lookup for
each entry's npm/PyPI package kicks off automatically, one
worker thread per entry, the moment the cards are built --
it is the one check a reviewer can't do by eye, so it must
never depend on a click. It fails soft (a dead registry
shows "unavailable", never blocks review or Sign) and a
per-card "Re-check" button covers manual retries. Every
changed entry must be individually acknowledged (its
checkbox ticked) before Sign unlocks. There is no
"acknowledge all" -- see catalog_review.py.
3. Sign -- re-fetches the current blob SHA and refuses to sign unless
it still matches the pinned SHA from step 1 (TOCTOU fix:
catalog_review.can_sign). On success, writes
data/catalog.json + data/catalog.json.sig and commits BOTH
in a single commit, then pushes -- so main is never red
between a catalog merge and its signature.
The signature must be the artefact of an actual review, not a step that
follows one. Signing IS the approval act.
"""
from __future__ import annotations
import argparse
import contextlib
import getpass
import html
import json
import re
import subprocess
import sys
import urllib.error
import urllib.request
from dataclasses import dataclass
from pathlib import Path
import bcc_core as core
import catalog_review as review
# --------------------------------------------------------------------------- #
# Constants
# --------------------------------------------------------------------------- #
GITEA_HOST = "git.avezzano.io"
GITEA_API_BASE = f"https://{GITEA_HOST}/api/v1"
REPO_OWNER = "the_og"
REPO_NAME = "better-claude-config"
CATALOG_PATH = "data/catalog.json"
SIG_PATH = "data/catalog.json.sig"
# Outside the repo, per issue #62 ("never committed, never plaintext"). A
# 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
# passphrase-encrypted (see catalog_review.encrypt_private_key), never raw.
KEY_STORAGE_DIR = Path.home() / ".bcc-catalog-console"
KEY_STORAGE_FILE = KEY_STORAGE_DIR / "signing_key.enc"
HTTP_TIMEOUT = 6.0
# --------------------------------------------------------------------------- #
# Key storage: OS keychain if available, else a passphrase-encrypted file
# outside the repo. Never plaintext, never an env var, never committed.
# --------------------------------------------------------------------------- #
def _keyring_module():
"""Best-effort import of the optional `keyring` package. Returns None if
it isn't installed -- this tool must work without it, falling back to
the encrypted-file path. `keyring` is deliberately NOT added to
requirements-dev.txt: this is a maintainer-only tool excluded from the
shipped app, so it doesn't need to justify a new runtime dependency for
every user the way bcc.py's dependencies do."""
try:
import keyring
return keyring
except ImportError:
return None
_KEYRING_SERVICE = "bcc-catalog-console"
_KEYRING_USERNAME = "signing-key"
def store_encrypted_key(blob: bytes) -> str:
"""Persist an already-encrypted key blob (see
catalog_review.encrypt_private_key). 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."""
keyring = _keyring_module()
if keyring is not None:
try:
keyring.set_password(_KEYRING_SERVICE, _KEYRING_USERNAME, blob.hex())
return "OS keychain (via the `keyring` package)"
except Exception:
pass # fall through to the file-based path
KEY_STORAGE_DIR.mkdir(parents=True, exist_ok=True)
KEY_STORAGE_FILE.write_bytes(blob)
with contextlib.suppress(OSError): # best-effort on platforms without POSIX perm bits
KEY_STORAGE_FILE.chmod(0o600)
return f"encrypted file at {KEY_STORAGE_FILE}"
def load_encrypted_key() -> bytes:
"""Load the encrypted key blob from wherever store_encrypted_key() put
it. Raises FileNotFoundError if no key has been generated yet."""
keyring = _keyring_module()
if keyring is not None:
try:
hex_blob = keyring.get_password(_KEYRING_SERVICE, _KEYRING_USERNAME)
if hex_blob:
return bytes.fromhex(hex_blob)
except Exception:
pass
if not KEY_STORAGE_FILE.exists():
raise FileNotFoundError(
f"No signing key found (checked the OS keychain and {KEY_STORAGE_FILE}). "
"Run `python catalog_console.py keygen` first."
)
return KEY_STORAGE_FILE.read_bytes()
def unlock_signing_key(passphrase: str) -> bytes:
"""Load + decrypt the signing key seed. Raises ValueError on a wrong
passphrase, FileNotFoundError if no key exists yet."""
blob = load_encrypted_key()
return review.decrypt_private_key(blob, passphrase)
# --------------------------------------------------------------------------- #
# git plumbing against a local clone. The clone's `origin` remote is assumed
# to already carry credentials (the "tokened remote" every other BCC
# maintainer script relies on) -- this module never handles a token itself.
# --------------------------------------------------------------------------- #
class GitError(RuntimeError):
pass
def _git(repo_dir: Path, *args: str, capture_bytes: bool = False):
cmd = ["git", "-C", str(repo_dir), *args]
result = subprocess.run(cmd, capture_output=True, check=False)
if result.returncode != 0:
stderr = result.stderr.decode("utf-8", "replace")
raise GitError(f"git {' '.join(args)} failed: {stderr}")
return result.stdout if capture_bytes else result.stdout.decode("utf-8", "replace")
def fetch_ref(repo_dir: Path, ref: str) -> str:
"""Fetch `ref` from origin and return the resulting commit SHA."""
_git(repo_dir, "fetch", "origin", ref)
return _git(repo_dir, "rev-parse", "FETCH_HEAD").strip()
def blob_sha_at(repo_dir: Path, commit: str, path: str) -> str:
"""The git blob SHA of `path` as it exists at `commit`. This is what
gets pinned at review-start and re-checked immediately before signing
(catalog_review.can_sign) -- the TOCTOU fix."""
return _git(repo_dir, "rev-parse", f"{commit}:{path}").strip()
def blob_bytes(repo_dir: Path, blob_sha: str) -> bytes:
return _git(repo_dir, "cat-file", "blob", blob_sha, capture_bytes=True)
def read_catalog_at_commit(repo_dir: Path, commit: str) -> tuple[bytes, str]:
"""Return (raw_bytes, blob_sha) for data/catalog.json at `commit`."""
sha = blob_sha_at(repo_dir, commit, CATALOG_PATH)
return blob_bytes(repo_dir, sha), sha
def commit_and_push_signed_catalog(
repo_dir: Path, raw_bytes: bytes, signature: bytes, *, branch: str = "main"
) -> str:
"""Write data/catalog.json + data/catalog.json.sig and commit BOTH in a
single commit, then push to `branch`. Returns the new commit SHA.
This is deliberate: if signing happened in a commit AFTER the catalog
merge, main would be red (payload present, signature missing) between
every catalog merge and its signing commit. Routine red-main trains
exactly the alarm fatigue this whole design exists to prevent. Emitting
one commit with both files means main is never in that state.
"""
_git(repo_dir, "checkout", branch)
_git(repo_dir, "pull", "--ff-only", "origin", branch)
(repo_dir / CATALOG_PATH).write_bytes(raw_bytes)
(repo_dir / SIG_PATH).write_bytes(signature)
_git(repo_dir, "add", CATALOG_PATH, SIG_PATH)
_git(
repo_dir,
"commit",
"-m",
"chore: sign data/catalog.json (Catalog Console, #62)\n\n"
"Payload and detached Ed25519 signature land together so main is "
"never red between a catalog merge and its signature.",
)
_git(repo_dir, "push", "origin", branch)
return _git(repo_dir, "rev-parse", "HEAD").strip()
# --------------------------------------------------------------------------- #
# Gitea REST API: list open PRs touching data/catalog.json
# --------------------------------------------------------------------------- #
def _gitea_get(path: str, token: str | None = None) -> object:
url = f"{GITEA_API_BASE}{path}"
req = urllib.request.Request(url)
if token:
req.add_header("Authorization", f"token {token}")
with urllib.request.urlopen(req, timeout=HTTP_TIMEOUT) as resp:
return json.loads(resp.read().decode("utf-8"))
@dataclass(frozen=True)
class CatalogPR:
number: int
title: str
head_ref: str # refs/pull/<n>/head
def list_open_catalog_prs(token: str | None = None) -> list[CatalogPR]:
"""Open PRs against REPO_OWNER/REPO_NAME whose diff touches
data/catalog.json. Fails soft: on any network error, returns [] rather
than raising into the GUI (Load still offers the `main` source)."""
try:
prs = _gitea_get(f"/repos/{REPO_OWNER}/{REPO_NAME}/pulls?state=open", token)
except (urllib.error.URLError, TimeoutError, ValueError):
return []
matches: list[CatalogPR] = []
for pr in prs or []:
number = pr.get("number")
if not isinstance(number, int):
continue
if _pr_touches_catalog(number, token):
matches.append(
CatalogPR(
number=number,
title=str(pr.get("title", f"PR #{number}")),
head_ref=f"refs/pull/{number}/head",
)
)
return matches
def _pr_touches_catalog(pr_number: int, token: str | None) -> bool:
url = f"https://{GITEA_HOST}/{REPO_OWNER}/{REPO_NAME}/pulls/{pr_number}.diff"
req = urllib.request.Request(url)
if token:
req.add_header("Authorization", f"token {token}")
try:
with urllib.request.urlopen(req, timeout=HTTP_TIMEOUT) as resp:
diff_text = resp.read().decode("utf-8", "replace")
except (urllib.error.URLError, TimeoutError):
return False
return CATALOG_PATH in diff_text
def token_from_git_remote(repo_dir: Path) -> str | None:
"""Best-effort extraction of a PAT embedded in `origin`'s URL
(https://<token>@host/...), matching the "tokened remote" every other
BCC maintainer flow already relies on. Returns None if there isn't one
(public read-only API calls still work, just rate-limited)."""
try:
url = _git(repo_dir, "remote", "get-url", "origin").strip()
except GitError:
return None
match = re.match(r"https://([^@/]+)@", url)
if not match:
return None
token = match.group(1)
# `user:token` form -- keep only the token half if present.
return token.split(":", 1)[-1]
# --------------------------------------------------------------------------- #
# Registry lookup fetchers (npm / PyPI). Kept out of catalog_review.py so the
# pure module never makes a network call itself -- these are injected as the
# `Fetcher` callable review.lookup_registry_info() expects.
# --------------------------------------------------------------------------- #
def fetch_npm_info(ref: review.PackageRef) -> dict | None:
url = f"https://registry.npmjs.org/{ref.name}"
try:
req = urllib.request.Request(url, headers={"Accept": "application/json"})
with urllib.request.urlopen(req, timeout=HTTP_TIMEOUT) as resp:
data = json.loads(resp.read().decode("utf-8"))
except (urllib.error.URLError, TimeoutError, ValueError):
return None
time_info = data.get("time") or {}
created = time_info.get("created")
modified = time_info.get("modified")
age_days = _iso_age_days(created)
maintainers = data.get("maintainers") or []
publisher = maintainers[0].get("name") if maintainers else None
downloads = None
try:
dl_url = f"https://api.npmjs.org/downloads/point/last-week/{ref.name}"
with urllib.request.urlopen(dl_url, timeout=HTTP_TIMEOUT) as resp:
downloads = json.loads(resp.read().decode("utf-8")).get("downloads")
except (urllib.error.URLError, TimeoutError, ValueError):
pass # fail soft -- downloads are a nice-to-have, not required
return {
"publisher": publisher,
"age_days": age_days,
"last_release": modified,
"downloads": downloads,
}
def fetch_pypi_info(ref: review.PackageRef) -> dict | None:
url = f"https://pypi.org/pypi/{ref.name}/json"
try:
with urllib.request.urlopen(url, timeout=HTTP_TIMEOUT) as resp:
data = json.loads(resp.read().decode("utf-8"))
except (urllib.error.URLError, TimeoutError, ValueError):
return None
info = data.get("info") or {}
releases = data.get("releases") or {}
last_release = None
earliest_upload = None
for files in releases.values():
for f in files:
uploaded = f.get("upload_time_iso_8601")
if not uploaded:
continue
if last_release is None or uploaded > last_release:
last_release = uploaded
if earliest_upload is None or uploaded < earliest_upload:
earliest_upload = uploaded
return {
"publisher": info.get("author") or info.get("maintainer"),
"age_days": _iso_age_days(earliest_upload),
"last_release": last_release,
"downloads": None, # PyPI JSON API doesn't include download counts
}
def _iso_age_days(iso_timestamp: str | None) -> int | None:
if not iso_timestamp:
return None
import datetime as _dt
try:
parsed = _dt.datetime.fromisoformat(iso_timestamp.replace("Z", "+00:00"))
now = _dt.datetime.now(_dt.timezone.utc)
return max((now - parsed).days, 0)
except ValueError:
return None
def registry_fetcher(ref: review.PackageRef) -> dict | None:
"""The Fetcher passed to review.lookup_registry_info(). Never raises --
both fetch_npm_info/fetch_pypi_info already fail soft, and
lookup_registry_info() wraps this in a try/except regardless."""
if ref.ecosystem == "npm":
return fetch_npm_info(ref)
if ref.ecosystem == "pypi":
return fetch_pypi_info(ref)
return None
# --------------------------------------------------------------------------- #
# GUI (PySide6). Everything above this line has no Qt dependency and is
# exercised by tests/test_catalog_review.py; everything below is a thin
# shell that calls into it.
# --------------------------------------------------------------------------- #
from PySide6.QtCore import Qt, QThread, Signal # noqa: E402
from PySide6.QtWidgets import ( # noqa: E402
QApplication,
QCheckBox,
QDialog,
QDialogButtonBox,
QFormLayout,
QGroupBox,
QHBoxLayout,
QLabel,
QLineEdit,
QListWidget,
QListWidgetItem,
QMainWindow,
QMessageBox,
QPushButton,
QScrollArea,
QVBoxLayout,
QWidget,
)
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), which means
an attacker-controlled description/notes/URL/package-name string
containing `<b>` or `<img onerror=...>` would render as markup instead
of visible text -- exactly the kind of thing that could hide a homoglyph
swap or make a risk warning easy to miss. Every catalog-derived string
shown by this Console MUST go through this helper (or otherwise set
Qt.PlainText explicitly) rather than a bare QLabel(...).
"""
label = QLabel(html.escape(str(text)))
label.setTextFormat(Qt.PlainText)
label.setWordWrap(True)
return label
_SEVERITY_PREFIX = {"blocking": "✖ BLOCKING", "warning": "⚠ WARNING", "info": "ℹ INFO"}
class RegistryLookupWorker(QThread):
"""Off-UI-thread registry lookups, mirroring bcc.py's ConnTester/
SpawnTester pattern. Never blocks the review UI on a slow/dead network."""
done = Signal(object) # list[review.RegistryInfo]
def __init__(self, refs: list[review.PackageRef], all_entry_ids: list[str]):
super().__init__()
self._refs = refs
self._all_entry_ids = all_entry_ids
def run(self):
results = [
review.lookup_registry_info(ref, registry_fetcher, self._all_entry_ids)
for ref in self._refs
]
self.done.emit(results)
class EntryCard(QWidget):
"""One changed catalog entry: the diff, risk findings, and the
acknowledge checkbox that gates Sign. `command`/`args` are rendered
visually dominant (bold-weight, larger, first) since they're the fields
that execute.
"""
acknowledged_changed = Signal(str, bool)
def __init__(self, change: review.EntryChange, all_entry_ids: list[str]):
super().__init__()
self.change = change
self._all_entry_ids = all_entry_ids
self._worker: RegistryLookupWorker | None = None
outline = QVBoxLayout(self)
box = QGroupBox(f"[{change.status.upper()}] {change.entry_id}")
outline.addWidget(box)
layout = QVBoxLayout(box)
entry = change.new or change.old or {}
config = entry.get("config") or {}
cmd_label = plain_label(f"command: {config.get('command', '(none)')}")
cmd_label.setStyleSheet("font-weight: bold; font-size: 13pt;")
layout.addWidget(cmd_label)
args_label = plain_label(f"args: {config.get('args', [])}")
args_label.setStyleSheet("font-weight: bold;")
layout.addWidget(args_label)
for fc in change.field_changes:
if fc.field in ("config.command", "config.args"):
continue # already shown dominant, above
layout.addWidget(plain_label(f"{fc.field}: {fc.old!r} -> {fc.new!r}"))
findings = review.entry_risk_findings(change)
for finding in findings:
prefix = _SEVERITY_PREFIX.get(finding.severity, finding.severity.upper())
flabel = plain_label(f"{prefix}: {finding.message}")
if finding.severity == "blocking":
flabel.setStyleSheet("color: #c62828; font-weight: bold;")
elif finding.severity == "warning":
flabel.setStyleSheet("color: #ef6c00;")
else:
flabel.setStyleSheet("color: #1565c0;")
layout.addWidget(flabel)
self.registry_label = plain_label("Registry lookup: loading...")
layout.addWidget(self.registry_label)
recheck_btn = QPushButton("Re-check")
recheck_btn.clicked.connect(self._run_registry_lookup)
layout.addWidget(recheck_btn)
self.blocking = any(f.severity == "blocking" for f in findings)
self.checkbox = QCheckBox(
"I have reviewed this entry, including command/args and the risk"
" annotations above, and approve it."
)
if self.blocking:
self.checkbox.setEnabled(False)
self.checkbox.setToolTip(
"This entry has a BLOCKING finding and cannot be acknowledged "
"until the underlying change is fixed (edit the PR, don't sign around it)."
)
self.checkbox.toggled.connect(
lambda checked: self.acknowledged_changed.emit(change.entry_id, checked)
)
layout.addWidget(self.checkbox)
# Registry lookup is the one check a reviewer can't do by eye -- it's
# what catches a typosquatted/hijacked package (it already caught
# firecrawl-mcp in the seed data). It must run automatically as soon
# as the card exists, not wait on a click a tired maintainer might
# skip at 11pm. Off the GUI thread (RegistryLookupWorker is a
# QThread) and fails soft: a dead/slow registry can never gate
# review or signing, it just leaves this entry's lookup showing
# "unavailable". The "Re-check" button above stays for retrying a
# failed/unavailable lookup by hand.
self._run_registry_lookup()
def _run_registry_lookup(self):
entry = self.change.new or {}
refs = review.extract_package_refs(entry)
if not refs:
self.registry_label.setText("Registry lookup: no npm/PyPI package in this entry.")
return
self.registry_label.setText("Registry lookup: loading...")
self._worker = RegistryLookupWorker(refs, self._all_entry_ids)
self._worker.done.connect(self._on_registry_result)
self._worker.start()
def _on_registry_result(self, results: list[review.RegistryInfo]):
lines = []
for info in results:
if not info.available:
lines.append(f"{info.ref.name}: unavailable (network/registry unreachable)")
continue
neighbor_note = (
f" | NEAR-NEIGHBOUR of: {', '.join(info.near_neighbor_ids)}"
if info.near_neighbor_ids
else ""
)
lines.append(
f"{info.ref.name}: publisher={info.publisher!r} age_days={info.age_days} "
f"last_release={info.last_release} downloads={info.downloads}{neighbor_note}"
)
text = "Registry lookup:\n" + "\n".join(lines)
self.registry_label.setText(html.escape(text))
self.registry_label.setTextFormat(Qt.PlainText)
class PassphraseDialog(QDialog):
def __init__(self, prompt: str, parent=None):
super().__init__(parent)
self.setWindowTitle("Signing key passphrase")
layout = QFormLayout(self)
self.edit = QLineEdit()
self.edit.setEchoMode(QLineEdit.EchoMode.Password)
layout.addRow(prompt, self.edit)
buttons = QDialogButtonBox(
QDialogButtonBox.StandardButton.Ok | QDialogButtonBox.StandardButton.Cancel
)
buttons.accepted.connect(self.accept)
buttons.rejected.connect(self.reject)
layout.addRow(buttons)
def passphrase(self) -> str:
return self.edit.text()
class ReviewWindow(QMainWindow):
def __init__(self, repo_dir: Path):
super().__init__()
self.repo_dir = repo_dir
self.session: review.ReviewSession | None = None
self.cards: dict[str, EntryCard] = {}
self.setWindowTitle("BCC Catalog Console -- maintainer-only, never shipped")
central = QWidget()
self.setCentralWidget(central)
root = QVBoxLayout(central)
top = QHBoxLayout()
self.source_list = QListWidget()
self.source_list.addItem(QListWidgetItem("main (current tip)"))
top.addWidget(self.source_list, 1)
side = QVBoxLayout()
load_btn = QPushButton("Load selected source")
load_btn.clicked.connect(self._on_load)
side.addWidget(load_btn)
refresh_prs_btn = QPushButton("Refresh open PR list")
refresh_prs_btn.clicked.connect(self._refresh_pr_list)
side.addWidget(refresh_prs_btn)
side.addStretch(1)
top.addLayout(side)
root.addLayout(top)
self.scroll = QScrollArea()
self.scroll.setWidgetResizable(True)
self.card_container = QWidget()
self.card_layout = QVBoxLayout(self.card_container)
self.scroll.setWidget(self.card_container)
root.addWidget(self.scroll, 1)
self.status_label = plain_label("Load a source to begin review.")
root.addWidget(self.status_label)
self.sign_btn = QPushButton("Sign")
self.sign_btn.setEnabled(False)
self.sign_btn.clicked.connect(self._on_sign)
root.addWidget(self.sign_btn)
self._token = token_from_git_remote(self.repo_dir)
self._prs: list[CatalogPR] = []
self._refresh_pr_list()
def _refresh_pr_list(self):
self._prs = list_open_catalog_prs(self._token)
while self.source_list.count() > 1:
self.source_list.takeItem(1)
for pr in self._prs:
self.source_list.addItem(QListWidgetItem(f"PR #{pr.number}: {pr.title}"))
def _on_load(self):
row = self.source_list.currentRow()
try:
if row <= 0:
commit = fetch_ref(self.repo_dir, "main")
old_commit = None # main vs itself has no "old" -- nothing to diff without a base
else:
pr = self._prs[row - 1]
commit = fetch_ref(self.repo_dir, pr.head_ref)
old_commit = fetch_ref(self.repo_dir, "main")
new_raw, new_blob_sha = read_catalog_at_commit(self.repo_dir, commit)
new_catalog = core.load_catalog(new_raw)
if old_commit:
old_raw, _old_sha = read_catalog_at_commit(self.repo_dir, old_commit)
old_catalog = core.load_catalog(old_raw)
else:
old_catalog = new_catalog
except (GitError, ValueError) as e:
QMessageBox.critical(self, "Load failed", html.escape(str(e)))
return
self._new_raw = new_raw
self.session = review.start_review(new_blob_sha, old_catalog, new_catalog)
self._render_cards()
def _render_cards(self):
while self.card_layout.count():
item = self.card_layout.takeAt(0)
if item.widget():
item.widget().deleteLater()
self.cards.clear()
assert self.session is not None
all_ids = sorted(
{e.get("id") for e in (self.session.new_catalog.get("servers") or []) if e.get("id")}
)
for change in self.session.changes:
card = EntryCard(change, all_ids)
card.acknowledged_changed.connect(self._on_acknowledge_changed)
self.cards[change.entry_id] = card
self.card_layout.addWidget(card)
self.card_layout.addStretch(1)
self._update_status()
def _on_acknowledge_changed(self, entry_id: str, checked: bool):
assert self.session is not None
if checked:
review.acknowledge_entry(self.session, entry_id)
else:
review.unacknowledge_entry(self.session, entry_id)
self._update_status()
def _update_status(self):
assert self.session is not None
all_ack = review.all_entries_acknowledged(self.session)
self.sign_btn.setEnabled(all_ack)
pending = len(self.session.changes) - len(self.session.acknowledged)
self.status_label.setText(
f"{len(self.session.changes)} changed entries, {pending} not yet acknowledged."
)
def _on_sign(self):
assert self.session is not None
try:
current_sha = blob_sha_at(self.repo_dir, fetch_ref(self.repo_dir, "main"), CATALOG_PATH)
except GitError as e:
QMessageBox.critical(self, "Sign failed", html.escape(str(e)))
return
decision = review.can_sign(self.session, current_sha)
if not decision.ok:
QMessageBox.warning(self, "Cannot sign", html.escape(decision.reason or ""))
if decision.reason and "changed" in decision.reason.lower():
self._on_load() # force a re-review against the new bytes
return
dialog = PassphraseDialog("Enter signing key passphrase:", self)
if dialog.exec() != QDialog.DialogCode.Accepted:
return
try:
seed = unlock_signing_key(dialog.passphrase())
except (FileNotFoundError, ValueError) as e:
QMessageBox.critical(self, "Sign failed", html.escape(str(e)))
return
signature = review.sign_catalog_bytes(self._new_raw, seed)
try:
new_commit = commit_and_push_signed_catalog(self.repo_dir, self._new_raw, signature)
except GitError as e:
QMessageBox.critical(self, "Commit/push failed", html.escape(str(e)))
return
QMessageBox.information(self, "Signed", f"Signed and pushed as commit {new_commit[:12]}.")
self.sign_btn.setEnabled(False)
# --------------------------------------------------------------------------- #
# CLI
# --------------------------------------------------------------------------- #
def cmd_keygen(_args: argparse.Namespace) -> int:
seed, pubkey = review.generate_keypair()
passphrase = getpass.getpass("Choose a passphrase to encrypt the new signing key: ")
confirm = getpass.getpass("Confirm passphrase: ")
if passphrase != confirm:
print("error: passphrases did not match", file=sys.stderr)
return 1
if not passphrase:
print("error: a non-empty passphrase is required", file=sys.stderr)
return 1
blob = review.encrypt_private_key(seed, passphrase)
where = store_encrypted_key(blob)
pubkey_b64 = __import__("base64").b64encode(pubkey).decode("ascii")
print(f"Private key encrypted and stored in: {where}")
print()
print("Public key (base64, paste into bcc_core.CATALOG_PUBKEYS):")
print(f" {pubkey_b64}")
print()
print(
"Also add it as the Gitea repo secret RELEASE_SIGNING_KEY (base64 of the "
"32-byte private seed) used by release.yml -- get that value with:"
)
print(" python catalog_console.py show-seed-b64 # careful: prints the raw key")
return 0
def cmd_show_seed_b64(_args: argparse.Namespace) -> int:
passphrase = getpass.getpass("Signing key passphrase: ")
try:
seed = unlock_signing_key(passphrase)
except (FileNotFoundError, ValueError) as e:
print(f"error: {e}", file=sys.stderr)
return 1
import base64
print(base64.b64encode(seed).decode("ascii"))
return 0
def cmd_gui(args: argparse.Namespace) -> int:
repo_dir = Path(args.repo).resolve()
if not (repo_dir / CATALOG_PATH).exists():
print(
f"error: {repo_dir} doesn't look like a BCC checkout (no {CATALOG_PATH})",
file=sys.stderr,
)
return 1
app = QApplication(sys.argv)
app.setApplicationName("BCC Catalog Console")
win = ReviewWindow(repo_dir)
win.resize(900, 700)
win.show()
return app.exec()
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description=__doc__)
sub = parser.add_subparsers(dest="command")
p_gui = sub.add_parser("gui", help="launch the review/sign GUI (default)")
p_gui.add_argument("--repo", default=".", help="path to a BCC git checkout (default: cwd)")
p_gui.set_defaults(func=cmd_gui)
p_keygen = sub.add_parser("keygen", help="generate a new Ed25519 signing keypair")
p_keygen.set_defaults(func=cmd_keygen)
p_seed = sub.add_parser(
"show-seed-b64", help="print the base64 private seed (for the RELEASE_SIGNING_KEY secret)"
)
p_seed.set_defaults(func=cmd_show_seed_b64)
return parser
_SUBCOMMANDS = ("gui", "keygen", "show-seed-b64", "-h", "--help")
def main(argv: list[str] | None = None) -> int:
argv = sys.argv[1:] if argv is None else list(argv)
# `python catalog_console.py` with no subcommand (or with GUI-only flags
# like --repo) launches the GUI -- "gui" is the default action.
if not argv or argv[0] not in _SUBCOMMANDS:
argv = ["gui", *argv]
parser = build_parser()
args = parser.parse_args(argv)
return args.func(args)
if __name__ == "__main__":
raise SystemExit(main())
catalog_review.py
+727
View File
@@ -0,0 +1,727 @@
"""
catalog_review.py -- pure, GUI-free review/diff/risk/crypto logic for the
Catalog Console (issue #62).
This module is deliberately Qt-free and network-free so every function in it
is unit-testable offline, exactly like bcc_core.py. catalog_console.py (the
PySide6 GUI) is a thin shell over these functions -- it owns Qt widgets,
subprocess/git calls, and HTTP registry lookups; this module owns judgment.
Nothing here is reimplemented from bcc_core: the command allowlist and the
signature domain-separation prefix are imported, not retyped, so the two
modules cannot silently drift apart (see bcc_core.validate_catalog /
bcc_core.verify_catalog_signature and the project's "the check drifted on a
new surface" recurring-bug lesson).
"""
from __future__ import annotations
import os
import re
from collections.abc import Callable
from dataclasses import dataclass, field
from urllib.parse import urlsplit
from bcc_core import _CATALOG_SIG_DOMAIN as CATALOG_SIG_DOMAIN
from bcc_core import CATALOG_ALLOWED_COMMANDS
# --------------------------------------------------------------------------- #
# Semantic diff
# --------------------------------------------------------------------------- #
# Top-level scalar/simple fields compared directly (not drilled into).
_DIFF_FIELDS = (
"display",
"description",
"category",
"official",
"setup",
"homepage",
"docs_url",
"source",
"notes",
"stars",
"last_release",
"env_required",
)
@dataclass(frozen=True)
class FieldChange:
"""One field that differs between the old and new version of an entry."""
field: str
old: object
new: object
@dataclass(frozen=True)
class EntryChange:
"""One catalog entry's change: added, removed, or changed.
`old`/`new` are the raw entry dicts (or None for added/removed) so risk
predicates and the GUI can inspect anything not captured by
`field_changes` (which only lists fields that actually differ).
"""
entry_id: str
status: str # "added" | "removed" | "changed"
old: dict | None
new: dict | None
field_changes: tuple[FieldChange, ...] = ()
def _config_field_changes(old_cfg: dict | None, new_cfg: dict | None) -> list[FieldChange]:
old_cfg = old_cfg or {}
new_cfg = new_cfg or {}
changes: list[FieldChange] = []
for f in ("command", "args", "env"):
ov, nv = old_cfg.get(f), new_cfg.get(f)
if ov != nv:
changes.append(FieldChange(f"config.{f}", ov, nv))
return changes
def _entry_field_changes(old_entry: dict, new_entry: dict) -> tuple[FieldChange, ...]:
changes: list[FieldChange] = []
for f in _DIFF_FIELDS:
ov, nv = old_entry.get(f), new_entry.get(f)
if ov != nv:
changes.append(FieldChange(f, ov, nv))
changes.extend(_config_field_changes(old_entry.get("config"), new_entry.get("config")))
return tuple(changes)
def diff_catalogs(old: dict | None, new: dict | None) -> list[EntryChange]:
"""Semantic (per-entry) diff between two parsed catalog dicts.
NOT a text diff: entries are matched by `id`, and each changed entry
reports exactly which fields differ (with before/after values), which is
what lets the Console render "command changed from X to Y" instead of a
JSON line diff a reviewer has to mentally reconstruct.
Entries missing/malformed `id` are ignored here -- that is a
validate_catalog() rejection, not a diffing concern, and diffing must not
silently invent a match for two differently-broken entries.
"""
old_servers = {
e["id"]: e
for e in (old or {}).get("servers", []) or []
if isinstance(e, dict) and isinstance(e.get("id"), str) and e.get("id")
}
new_servers = {
e["id"]: e
for e in (new or {}).get("servers", []) or []
if isinstance(e, dict) and isinstance(e.get("id"), str) and e.get("id")
}
changes: list[EntryChange] = []
for entry_id in sorted(set(old_servers) | set(new_servers)):
old_e = old_servers.get(entry_id)
new_e = new_servers.get(entry_id)
if old_e is None:
changes.append(EntryChange(entry_id, "added", None, new_e, ()))
elif new_e is None:
changes.append(EntryChange(entry_id, "removed", old_e, None, ()))
elif old_e != new_e:
fc = _entry_field_changes(old_e, new_e)
if fc:
changes.append(EntryChange(entry_id, "changed", old_e, new_e, fc))
return changes
# --------------------------------------------------------------------------- #
# Risk annotations -- each predicate is pure and independently unit-tested.
# --------------------------------------------------------------------------- #
@dataclass(frozen=True)
class RiskFinding:
severity: str # "blocking" | "warning" | "info"
code: str
message: str
def _escape_non_ascii(s: str) -> str:
"""Render a string with any non-ASCII code point shown as an escape
sequence, so a homoglyph/RTL-override character can't visually pass as
the real thing in the review UI."""
return s.encode("unicode_escape").decode("ascii")
def risk_env_required(change: EntryChange) -> list[RiskFinding]:
"""A non-empty env_required value is blocking: catalog entries must ship
only the *names* of env vars the user fills in, never values."""
entry = change.new or {}
env_required = entry.get("env_required")
findings: list[RiskFinding] = []
if isinstance(env_required, dict):
for k, v in env_required.items():
if v not in (None, ""):
findings.append(
RiskFinding(
"blocking",
"env_required_value",
f"env_required[{k!r}] carries a non-empty value -- catalog "
"entries must never ship secret values, only placeholder names.",
)
)
return findings
def risk_command_allowlist(change: EntryChange) -> list[RiskFinding]:
"""A command outside bcc_core.CATALOG_ALLOWED_COMMANDS is blocking.
Imports the allowlist rather than redefining it."""
entry = change.new or {}
config = entry.get("config") or {}
command = config.get("command")
if isinstance(command, str) and command and command not in CATALOG_ALLOWED_COMMANDS:
return [
RiskFinding(
"blocking",
"command_not_allowed",
f"command {command!r} is not on the catalog allowlist "
f"({', '.join(sorted(CATALOG_ALLOWED_COMMANDS))}).",
)
]
return []
def risk_non_ascii(change: EntryChange) -> list[RiskFinding]:
"""Non-ASCII code points in id/command/args are blocking -- homoglyph /
RTL-override typosquatting can make a malicious package name visually
identical to a legitimate one in a naive diff view."""
entry = change.new or {}
findings: list[RiskFinding] = []
entry_id = entry.get("id")
if isinstance(entry_id, str) and not entry_id.isascii():
findings.append(
RiskFinding(
"blocking",
"non_ascii_id",
f"id contains non-ASCII code points: {_escape_non_ascii(entry_id)!r}",
)
)
config = entry.get("config") or {}
command = config.get("command")
if isinstance(command, str) and not command.isascii():
findings.append(
RiskFinding(
"blocking",
"non_ascii_command",
f"command contains non-ASCII code points: {_escape_non_ascii(command)!r}",
)
)
for a in config.get("args") or []:
if isinstance(a, str) and not a.isascii():
findings.append(
RiskFinding(
"blocking",
"non_ascii_arg",
f"arg contains non-ASCII code points: {_escape_non_ascii(a)!r}",
)
)
return findings
def _npm_candidate_args(command: str | None, args: list[str]) -> list[str]:
if command != "npx":
return []
return [a for a in args if isinstance(a, str) and a and not a.startswith("-")]
def split_npm_spec(spec: str) -> tuple[str, str | None]:
"""Split an npm package spec into (name, version). version is None if
unpinned. Handles scoped (@scope/name@version) and unscoped
(name@version) specs."""
if spec.startswith("@"):
rest = spec[1:]
if "/" not in rest:
return spec, None # malformed scope, can't tell -- treat unpinned
scope, _, remainder = rest.partition("/")
if "@" in remainder:
pkg_name, _, version = remainder.partition("@")
return f"@{scope}/{pkg_name}", (version or None)
return f"@{scope}/{remainder}", None
if "@" in spec:
name, _, version = spec.partition("@")
return name, (version or None)
return spec, None
def is_pinned_npm_spec(spec: str) -> bool:
_name, version = split_npm_spec(spec)
return bool(version)
_DOCKER_VALUE_FLAGS = {
"-e",
"--env",
"-v",
"--volume",
"-p",
"--publish",
"-w",
"--workdir",
"-u",
"--user",
"--name",
"--network",
"--entrypoint",
}
def docker_image_candidates(args: list[str]) -> list[str]:
"""Best-effort extraction of the image reference from a `docker run
[OPTIONS] IMAGE [CMD...]` args list: the first positional token after
any leading `run` and flag(+value) pairs."""
candidates: list[str] = []
i = 0
while i < len(args):
a = args[i]
if a == "run":
i += 1
continue
if isinstance(a, str) and a.startswith("-"):
if "=" not in a and a in _DOCKER_VALUE_FLAGS:
i += 2
continue
i += 1
continue
if isinstance(a, str):
candidates.append(a)
break # first positional token after `run` is the image ref
return candidates
def is_pinned_docker_image(image: str) -> bool:
if "@sha256:" in image:
return True
tag_part = image.rsplit("/", 1)[-1]
if ":" not in tag_part:
return False # no tag => implicit :latest
tag = tag_part.rsplit(":", 1)[-1]
return bool(tag) and tag != "latest"
def risk_unpinned_package(change: EntryChange) -> list[RiskFinding]:
"""Every entry must pin an exact version: an `@scope/pkg` npm arg with
no `@version`, or a docker image with no tag / `:latest`, is blocking.
A later-compromised package must not be able to auto-upgrade into every
user just because the catalog entry never pinned a version."""
entry = change.new or {}
config = entry.get("config") or {}
command = config.get("command")
args = config.get("args") or []
findings: list[RiskFinding] = []
if command == "npx":
for a in _npm_candidate_args(command, args):
if not is_pinned_npm_spec(a):
findings.append(
RiskFinding(
"blocking",
"unpinned_npm_package",
f"npm package arg {a!r} has no pinned @version.",
)
)
elif command == "docker":
for img in docker_image_candidates(args):
if not is_pinned_docker_image(img):
findings.append(
RiskFinding(
"blocking",
"unpinned_docker_image",
f"docker image {img!r} is not pinned to an exact tag "
"(uses :latest or no tag).",
)
)
return findings
_URL_FIELDS = ("homepage", "docs_url", "source")
def _domain(url: str) -> str:
try:
return urlsplit(url).netloc.lower()
except ValueError:
return ""
def risk_url_domain_change(change: EntryChange) -> list[RiskFinding]:
"""Non-https URLs and, more importantly, a *domain change* on any URL
field are surfaced loudly with old-vs-new domains broken out -- the
lookalike-domain-swap defence."""
findings: list[RiskFinding] = []
old_entry = change.old or {}
new_entry = change.new or {}
for f in _URL_FIELDS:
new_url = new_entry.get(f)
if not isinstance(new_url, str) or not new_url:
continue
if not new_url.startswith("https://"):
findings.append(
RiskFinding("warning", "non_https_url", f"{f} is not https://: {new_url!r}")
)
old_url = old_entry.get(f)
if isinstance(old_url, str) and old_url:
old_domain, new_domain = _domain(old_url), _domain(new_url)
if old_domain and new_domain and old_domain != new_domain:
findings.append(
RiskFinding(
"warning",
"domain_changed",
f"{f} domain changed from {old_domain!r} to {new_domain!r} -- "
"verify this isn't a lookalike-domain swap.",
)
)
return findings
def risk_new_entry(change: EntryChange) -> list[RiskFinding]:
"""A brand-new entry is flagged for extra scrutiny -- not blocking on its
own, but it's the category of change the registry lookup exists for."""
if change.status == "added":
return [
RiskFinding(
"info",
"new_entry",
"Brand-new catalog entry -- extra scrutiny: check publisher identity "
"via the registry lookup before signing.",
)
]
return []
_RISK_PREDICATES: tuple[Callable[[EntryChange], list[RiskFinding]], ...] = (
risk_env_required,
risk_command_allowlist,
risk_non_ascii,
risk_unpinned_package,
risk_url_domain_change,
risk_new_entry,
)
def entry_risk_findings(change: EntryChange) -> list[RiskFinding]:
"""Run every risk predicate against one entry change and return the
combined findings (order matches _RISK_PREDICATES)."""
findings: list[RiskFinding] = []
for predicate in _RISK_PREDICATES:
findings.extend(predicate(change))
return findings
def has_blocking_risk(change: EntryChange) -> bool:
return any(f.severity == "blocking" for f in entry_risk_findings(change))
# --------------------------------------------------------------------------- #
# Review session: acknowledge-gating + TOCTOU blob-SHA pinning
# --------------------------------------------------------------------------- #
@dataclass
class ReviewSession:
"""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
can_sign()."""
pinned_blob_sha: str
old_catalog: dict
new_catalog: dict
changes: list[EntryChange] = field(default_factory=list)
acknowledged: set[str] = field(default_factory=set)
def __post_init__(self) -> None:
if not self.changes:
self.changes = diff_catalogs(self.old_catalog, self.new_catalog)
def start_review(pinned_blob_sha: str, old_catalog: dict, new_catalog: dict) -> ReviewSession:
return ReviewSession(
pinned_blob_sha=pinned_blob_sha, old_catalog=old_catalog, new_catalog=new_catalog
)
def acknowledge_entry(session: ReviewSession, entry_id: str) -> None:
ids = {c.entry_id for c in session.changes}
if entry_id not in ids:
raise ValueError(f"{entry_id!r} is not part of this review session's diff.")
session.acknowledged.add(entry_id)
def unacknowledge_entry(session: ReviewSession, entry_id: str) -> None:
session.acknowledged.discard(entry_id)
def all_entries_acknowledged(session: ReviewSession) -> bool:
return {c.entry_id for c in session.changes} <= session.acknowledged
# NOTE for future editors: do NOT add an "acknowledge all" shortcut here, now
# or ever. The friction of individually acknowledging every changed entry is
# the entire point of this tool (issue #62) -- a shortcut would let a tired
# reviewer rubber-stamp a diff exactly like the "merge PR, run script, push"
# reflex this Console exists to replace. If this comment is the only thing
# stopping you, that is the point: it is stopping you on purpose.
@dataclass(frozen=True)
class SignDecision:
ok: bool
reason: str | None = None
def can_sign(session: ReviewSession, current_blob_sha: str) -> SignDecision:
"""Whether the Sign button may fire right now.
Two independent gates, both required:
1. TOCTOU: `current_blob_sha` (fetched fresh, immediately before signing)
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
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"
true rather than aspirational: the signature is bound to the exact
reviewed bytes, not to "whatever the file happens to be now".
2. Every changed entry in the diff must be individually acknowledged.
"""
if current_blob_sha != session.pinned_blob_sha:
return SignDecision(
False,
"The reviewed bytes changed since this review began (blob SHA "
"mismatch) -- re-review required before signing.",
)
if not all_entries_acknowledged(session):
pending = sorted({c.entry_id for c in session.changes} - session.acknowledged)
return SignDecision(
False, f"Not every changed entry has been acknowledged yet: {', '.join(pending)}"
)
return SignDecision(True, None)
def catalog_signing_message(raw_bytes: bytes) -> bytes:
"""The exact bytes that get signed: bcc_core's domain-separation prefix
(imported, never retyped) + the raw catalog bytes. Using this function
guarantees the Console's signature and bcc_core.verify_catalog_signature
can never drift apart on the prefix."""
return CATALOG_SIG_DOMAIN + raw_bytes
# --------------------------------------------------------------------------- #
# Key management: passphrase-encrypted-at-rest Ed25519 seed
#
# The private key is NEVER stored plaintext, never an env var, never
# committed. encrypt_private_key/decrypt_private_key are pure and offline
# (scrypt KDF + AES-256-GCM via `cryptography`, already a project
# dependency); catalog_console.py decides WHERE the resulting blob lives
# (OS keychain if available, else a file outside the repo).
# --------------------------------------------------------------------------- #
_KDF_SALT_LEN = 16
_KDF_N = 2**15 # scrypt cost parameter, tuned for a one-off interactive unlock
_KDF_R = 8
_KDF_P = 1
_NONCE_LEN = 12
_AAD = b"bcc-catalog-console-key-v1"
def _derive_key(passphrase: str, salt: bytes) -> bytes:
from cryptography.hazmat.primitives.kdf.scrypt import Scrypt
kdf = Scrypt(salt=salt, length=32, n=_KDF_N, r=_KDF_R, p=_KDF_P)
return kdf.derive(passphrase.encode("utf-8"))
def generate_keypair() -> tuple[bytes, bytes]:
"""Generate a new Ed25519 keypair. Returns (seed_32_bytes, pubkey_32_bytes)."""
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat
private_key = Ed25519PrivateKey.generate()
seed = private_key.private_bytes_raw()
pubkey = private_key.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw)
return seed, pubkey
def encrypt_private_key(seed: bytes, passphrase: str) -> bytes:
"""Encrypt a 32-byte Ed25519 seed at rest with a passphrase. Returns a
self-contained blob: salt || nonce || ciphertext+tag."""
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
if len(seed) != 32:
raise ValueError(f"expected a 32-byte raw Ed25519 seed, got {len(seed)} bytes")
if not passphrase:
raise ValueError("a non-empty passphrase is required")
salt = os.urandom(_KDF_SALT_LEN)
key = _derive_key(passphrase, salt)
nonce = os.urandom(_NONCE_LEN)
ciphertext = AESGCM(key).encrypt(nonce, seed, _AAD)
return salt + nonce + ciphertext
def decrypt_private_key(blob: bytes, passphrase: str) -> bytes:
"""Decrypt a blob produced by encrypt_private_key. Raises ValueError on a
wrong passphrase or corrupt blob -- never silently returns garbage."""
from cryptography.exceptions import InvalidTag
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
if len(blob) < _KDF_SALT_LEN + _NONCE_LEN:
raise ValueError("key blob is too short to be valid")
salt = blob[:_KDF_SALT_LEN]
nonce = blob[_KDF_SALT_LEN : _KDF_SALT_LEN + _NONCE_LEN]
ciphertext = blob[_KDF_SALT_LEN + _NONCE_LEN :]
key = _derive_key(passphrase, salt)
try:
return AESGCM(key).decrypt(nonce, ciphertext, _AAD)
except InvalidTag as e:
raise ValueError("wrong passphrase or corrupted key file") from e
def sign_catalog_bytes(raw: bytes, seed: bytes) -> bytes:
"""Sign `raw` catalog bytes with a 32-byte Ed25519 seed, using the exact
domain-separated message bcc_core.verify_catalog_signature expects."""
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
if len(seed) != 32:
raise ValueError(f"expected a 32-byte raw Ed25519 seed, got {len(seed)} bytes")
private_key = Ed25519PrivateKey.from_private_bytes(seed)
return private_key.sign(catalog_signing_message(raw))
# --------------------------------------------------------------------------- #
# Registry lookup -- the check a human genuinely can't do.
#
# The network call itself is injected (a `Fetcher` callable) so this stays
# testable offline; catalog_console.py supplies the real npm/PyPI HTTP
# fetcher. Fails soft everywhere: a fetcher returning None/raising just
# yields RegistryInfo(available=False), never an exception into the caller
# and never a block on review.
# --------------------------------------------------------------------------- #
@dataclass(frozen=True)
class PackageRef:
entry_id: str
ecosystem: str # "npm" | "pypi"
name: str
version: str | None
def split_pypi_spec(spec: str) -> tuple[str, str | None]:
for sep in ("==", "@"):
if sep in spec:
name, _, version = spec.partition(sep)
return name, (version or None)
return spec, None
def extract_package_refs(entry: dict) -> list[PackageRef]:
"""Pull out the package(s) a basic-tier entry's args reference, for the
registry lookup. Returns [] for link-only entries or entries whose
command isn't npx/uvx (docker images aren't registry-lookup candidates
in the npm/PyPI sense used here)."""
config = entry.get("config") or {}
command = config.get("command")
args = config.get("args") or []
entry_id = entry.get("id", "") if isinstance(entry.get("id"), str) else ""
refs: list[PackageRef] = []
if command == "npx":
for a in _npm_candidate_args(command, args):
name, version = split_npm_spec(a)
if name:
refs.append(PackageRef(entry_id, "npm", name, version))
elif command == "uvx":
for a in args:
if isinstance(a, str) and a and not a.startswith("-"):
name, version = split_pypi_spec(a)
if name:
refs.append(PackageRef(entry_id, "pypi", name, version))
break # `uvx <pkg>` -- first positional token is the package
return refs
@dataclass(frozen=True)
class RegistryInfo:
ref: PackageRef
available: bool
publisher: str | None = None
age_days: int | None = None
last_release: str | None = None
downloads: int | None = None
near_neighbor_ids: tuple[str, ...] = ()
Fetcher = Callable[[PackageRef], dict | None]
def edit_distance(a: str, b: str) -> int:
"""Levenshtein distance, iterative DP (no recursion depth concerns)."""
if a == b:
return 0
la, lb = len(a), len(b)
if la == 0:
return lb
if lb == 0:
return la
prev = list(range(lb + 1))
for i, ca in enumerate(a, 1):
cur = [i] + [0] * lb
for j, cb in enumerate(b, 1):
cost = 0 if ca == cb else 1
cur[j] = min(prev[j] + 1, cur[j - 1] + 1, prev[j - 1] + cost)
prev = cur
return prev[lb]
def near_neighbor_ids(name: str, other_ids: list[str], max_distance: int = 2) -> list[str]:
"""Catalog ids within `max_distance` edits of `name` (case-insensitive),
excluding an exact match -- the dependency-confusion / typosquat
near-neighbour warning."""
lname = name.lower()
return [
oid
for oid in other_ids
if oid != name and edit_distance(lname, oid.lower()) <= max_distance
]
def lookup_registry_info(
ref: PackageRef, fetcher: Fetcher, all_entry_ids: list[str]
) -> RegistryInfo:
"""Resolve one package against the live registry via the injected
fetcher. Never raises: any fetcher exception or falsy return means
`available=False` ("unavailable"), which the GUI renders plainly rather
than blocking or erroring the review."""
neighbors = tuple(near_neighbor_ids(ref.name, all_entry_ids))
try:
raw = fetcher(ref)
except Exception:
raw = None
if not raw:
return RegistryInfo(ref=ref, available=False, near_neighbor_ids=neighbors)
return RegistryInfo(
ref=ref,
available=True,
publisher=raw.get("publisher"),
age_days=raw.get("age_days"),
last_release=raw.get("last_release"),
downloads=raw.get("downloads"),
near_neighbor_ids=neighbors,
)
_NON_ASCII_RE = re.compile(r"[^\x00-\x7f]")
def contains_non_ascii(s: str) -> bool:
return bool(_NON_ASCII_RE.search(s))
tests/test_catalog_console_packaging.py
+66
View File
@@ -0,0 +1,66 @@
"""Asserts the maintainer-only Catalog Console (catalog_console.py,
catalog_review.py) is never bundled into the release binary.
A signing/review tool shipping to end users would be an own-goal (issue
#62): it has no reason to run on a user's machine, and its presence would
be a confusing artefact of a build that's supposed to be a thin GUI over
mcpServers config editing."""
from __future__ import annotations
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parent.parent
SPEC_PATH = REPO_ROOT / "bcc.spec"
_EXCLUDED_FILES = ("catalog_console.py", "catalog_review.py")
def test_spec_file_exists():
assert SPEC_PATH.exists()
def test_console_files_not_named_in_spec():
"""The spec text must never reference either maintainer-only module --
not as the Analysis entry point, not in datas, not anywhere."""
spec_text = SPEC_PATH.read_text(encoding="utf-8")
for filename in _EXCLUDED_FILES:
assert filename not in spec_text, (
f"{filename} must never be referenced by bcc.spec -- it is a "
"maintainer-only tool and must not ship to users."
)
def test_spec_analysis_entry_point_is_bcc_py_only():
"""PyInstaller's Analysis(...) call determines the dependency-scanned
entry point(s); it must be bcc.py alone."""
spec_text = SPEC_PATH.read_text(encoding="utf-8")
assert 'Analysis(\n ["bcc.py"],' in spec_text or 'Analysis(["bcc.py"]' in spec_text, (
"bcc.spec's Analysis(...) entry point changed shape -- re-verify by hand "
"that catalog_console.py / catalog_review.py are still excluded."
)
def test_console_modules_exist_but_are_standalone_top_level_files():
"""Sanity check the files this test is guarding actually exist as
top-level modules (not, say, silently moved into a package PyInstaller's
Analysis would still pick up as an implicit import of bcc.py)."""
for filename in _EXCLUDED_FILES:
assert (REPO_ROOT / filename).exists()
# bcc.py must not import them.
bcc_text = (REPO_ROOT / "bcc.py").read_text(encoding="utf-8")
module_name = filename.removesuffix(".py")
assert f"import {module_name}" not in bcc_text
assert f"from {module_name}" not in bcc_text
def test_requirements_files_do_not_reference_console_only_needs():
"""catalog_console.py's only import beyond the shipped stack is the
optional `keyring` package, which is intentionally NOT added as a hard
dependency anywhere a user install would pick it up."""
for req_file in ("requirements.txt", "requirements-dev.txt"):
path = REPO_ROOT / req_file
if not path.exists():
continue
text = path.read_text(encoding="utf-8").lower()
assert "keyring" not in text
tests/test_catalog_review.py
+555
View File
@@ -0,0 +1,555 @@
"""Tests for catalog_review.py -- semantic diff, risk predicates, review
session (acknowledge-gating + TOCTOU blob pinning), key encryption, and
registry-lookup logic for the Catalog Console (issue #62)."""
from __future__ import annotations
import sys
from pathlib import Path
import pytest
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import bcc_core as c
import catalog_review as r
def _entry(**overrides):
base = {
"id": "filesystem",
"display": "Filesystem",
"description": "desc",
"category": "files",
"homepage": "https://github.com/modelcontextprotocol/servers",
"stars": 100,
"official": True,
"setup": "basic",
"config": {
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-filesystem@1.0.0"],
},
"env_required": {},
"docs_url": "https://github.com/modelcontextprotocol/servers",
"notes": "",
"last_release": "2026-01-01",
}
base.update(overrides)
return base
def _catalog(*entries):
return {"schema": 1, "version": 1, "servers": list(entries)}
# --------------------------------------------------------------------------- #
# diff_catalogs
# --------------------------------------------------------------------------- #
def test_diff_detects_added_entry():
old = _catalog()
new = _catalog(_entry())
changes = r.diff_catalogs(old, new)
assert len(changes) == 1
assert changes[0].status == "added"
assert changes[0].entry_id == "filesystem"
assert changes[0].old is None
def test_diff_detects_removed_entry():
old = _catalog(_entry())
new = _catalog()
changes = r.diff_catalogs(old, new)
assert len(changes) == 1
assert changes[0].status == "removed"
assert changes[0].new is None
def test_diff_detects_no_change():
e = _entry()
old = _catalog(e)
new = _catalog(dict(e))
assert r.diff_catalogs(old, new) == []
def test_diff_detects_changed_command_and_args():
old = _catalog(_entry())
new = _catalog(_entry(config={"command": "uvx", "args": ["other-pkg@2.0.0"]}))
changes = r.diff_catalogs(old, new)
assert len(changes) == 1
ch = changes[0]
assert ch.status == "changed"
fields = {fc.field for fc in ch.field_changes}
assert "config.command" in fields
assert "config.args" in fields
def test_diff_detects_description_change():
old = _catalog(_entry())
new = _catalog(_entry(description="new description"))
changes = r.diff_catalogs(old, new)
assert changes[0].field_changes == (r.FieldChange("description", "desc", "new description"),)
def test_diff_ignores_entries_without_id():
old = _catalog()
new = _catalog({"display": "no id"})
assert r.diff_catalogs(old, new) == []
def test_diff_multiple_entries_sorted_by_id():
old = _catalog(_entry(id="zeta"), _entry(id="alpha"))
new = _catalog(
_entry(id="zeta", description="changed"), _entry(id="alpha", description="changed")
)
changes = r.diff_catalogs(old, new)
assert [c_.entry_id for c_ in changes] == ["alpha", "zeta"]
# --------------------------------------------------------------------------- #
# risk_env_required
# --------------------------------------------------------------------------- #
def test_risk_env_required_blocking_on_nonempty_value():
change = r.EntryChange("x", "changed", None, _entry(env_required={"API_KEY": "sk-real-value"}))
findings = r.risk_env_required(change)
assert len(findings) == 1
assert findings[0].severity == "blocking"
assert findings[0].code == "env_required_value"
def test_risk_env_required_clean_on_empty_value():
change = r.EntryChange("x", "changed", None, _entry(env_required={"API_KEY": ""}))
assert r.risk_env_required(change) == []
# --------------------------------------------------------------------------- #
# risk_command_allowlist
# --------------------------------------------------------------------------- #
def test_risk_command_allowlist_blocks_disallowed_command():
change = r.EntryChange(
"x", "changed", None, _entry(config={"command": "bash", "args": ["-c", "evil"]})
)
findings = r.risk_command_allowlist(change)
assert len(findings) == 1
assert findings[0].severity == "blocking"
def test_risk_command_allowlist_allows_listed_command():
for cmd in sorted(c.CATALOG_ALLOWED_COMMANDS):
change = r.EntryChange("x", "changed", None, _entry(config={"command": cmd, "args": []}))
assert r.risk_command_allowlist(change) == []
# --------------------------------------------------------------------------- #
# risk_non_ascii
# --------------------------------------------------------------------------- #
def test_risk_non_ascii_flags_homoglyph_id():
# Cyrillic 'а' (U+0430) instead of Latin 'a' -- classic homoglyph swap.
evil_id = "filаsystem"
change = r.EntryChange(evil_id, "changed", None, _entry(id=evil_id))
findings = r.risk_non_ascii(change)
assert any(f.code == "non_ascii_id" for f in findings)
assert findings[0].severity == "blocking"
# the offending string must be rendered with escapes, not raw
assert "\\u0430" in findings[0].message
def test_risk_non_ascii_flags_arg():
change = r.EntryChange(
"x", "changed", None, _entry(config={"command": "npx", "args": ["pаckage@1.0.0"]})
)
findings = r.risk_non_ascii(change)
assert any(f.code == "non_ascii_arg" for f in findings)
def test_risk_non_ascii_clean_for_ascii_entry():
change = r.EntryChange("x", "changed", None, _entry())
assert r.risk_non_ascii(change) == []
# --------------------------------------------------------------------------- #
# risk_unpinned_package
# --------------------------------------------------------------------------- #
def test_risk_unpinned_npm_package_no_version():
change = r.EntryChange(
"x",
"changed",
None,
_entry(config={"command": "npx", "args": ["-y", "@scope/pkg"]}),
)
findings = r.risk_unpinned_package(change)
assert len(findings) == 1
assert findings[0].code == "unpinned_npm_package"
assert findings[0].severity == "blocking"
def test_risk_pinned_npm_package_is_clean():
change = r.EntryChange(
"x",
"changed",
None,
_entry(config={"command": "npx", "args": ["-y", "@scope/pkg@1.2.3"]}),
)
assert r.risk_unpinned_package(change) == []
def test_risk_unpinned_unscoped_npm_package():
change = r.EntryChange(
"x", "changed", None, _entry(config={"command": "npx", "args": ["-y", "somepkg"]})
)
findings = r.risk_unpinned_package(change)
assert len(findings) == 1
def test_risk_unpinned_docker_latest_tag():
change = r.EntryChange(
"x",
"changed",
None,
_entry(config={"command": "docker", "args": ["run", "-i", "--rm", "myimage:latest"]}),
)
findings = r.risk_unpinned_package(change)
assert len(findings) == 1
assert findings[0].code == "unpinned_docker_image"
def test_risk_unpinned_docker_no_tag():
change = r.EntryChange(
"x", "changed", None, _entry(config={"command": "docker", "args": ["run", "myimage"]})
)
findings = r.risk_unpinned_package(change)
assert len(findings) == 1
def test_risk_pinned_docker_image_is_clean():
change = r.EntryChange(
"x",
"changed",
None,
_entry(config={"command": "docker", "args": ["run", "-i", "--rm", "myimage:1.2.3"]}),
)
assert r.risk_unpinned_package(change) == []
def test_risk_docker_digest_pin_is_clean():
change = r.EntryChange(
"x",
"changed",
None,
_entry(
config={
"command": "docker",
"args": ["run", "myimage@sha256:" + "a" * 64],
}
),
)
assert r.risk_unpinned_package(change) == []
# --------------------------------------------------------------------------- #
# risk_url_domain_change
# --------------------------------------------------------------------------- #
def test_risk_url_domain_change_warns_on_lookalike_swap():
old_entry = _entry(homepage="https://github.com/foo/bar")
new_entry = _entry(homepage="https://githu6.com/foo/bar")
change = r.EntryChange("x", "changed", old_entry, new_entry)
findings = r.risk_url_domain_change(change)
assert any(f.code == "domain_changed" for f in findings)
domain_finding = next(f for f in findings if f.code == "domain_changed")
assert "github.com" in domain_finding.message
assert "githu6.com" in domain_finding.message
assert domain_finding.severity == "warning"
def test_risk_url_domain_change_clean_when_domain_unchanged():
old_entry = _entry(homepage="https://github.com/foo/bar")
new_entry = _entry(homepage="https://github.com/foo/bar-renamed")
change = r.EntryChange("x", "changed", old_entry, new_entry)
assert r.risk_url_domain_change(change) == []
def test_risk_url_non_https_warns():
new_entry = _entry(homepage="http://example.com")
change = r.EntryChange("x", "changed", None, new_entry)
findings = r.risk_url_domain_change(change)
assert any(f.code == "non_https_url" for f in findings)
# --------------------------------------------------------------------------- #
# risk_new_entry / entry_risk_findings / has_blocking_risk
# --------------------------------------------------------------------------- #
def test_risk_new_entry_flags_added():
change = r.EntryChange("x", "added", None, _entry())
findings = r.risk_new_entry(change)
assert len(findings) == 1
assert findings[0].severity == "info"
def test_risk_new_entry_silent_for_changed():
change = r.EntryChange("x", "changed", _entry(), _entry(description="x"))
assert r.risk_new_entry(change) == []
def test_has_blocking_risk_true_for_disallowed_command():
change = r.EntryChange("x", "changed", None, _entry(config={"command": "bash", "args": []}))
assert r.has_blocking_risk(change) is True
def test_has_blocking_risk_false_for_clean_entry():
change = r.EntryChange("x", "changed", _entry(), _entry(description="new"))
assert r.has_blocking_risk(change) is False
# --------------------------------------------------------------------------- #
# ReviewSession: acknowledge gating
# --------------------------------------------------------------------------- #
def test_start_review_computes_diff():
old = _catalog()
new = _catalog(_entry())
session = r.start_review("sha1", old, new)
assert len(session.changes) == 1
def test_all_entries_acknowledged_false_initially():
session = r.start_review("sha1", _catalog(), _catalog(_entry()))
assert r.all_entries_acknowledged(session) is False
def test_acknowledge_entry_marks_acknowledged():
session = r.start_review("sha1", _catalog(), _catalog(_entry()))
r.acknowledge_entry(session, "filesystem")
assert r.all_entries_acknowledged(session) is True
def test_acknowledge_unknown_entry_raises():
session = r.start_review("sha1", _catalog(), _catalog(_entry()))
with pytest.raises(ValueError):
r.acknowledge_entry(session, "not-in-diff")
def test_acknowledge_gating_requires_every_entry():
session = r.start_review("sha1", _catalog(), _catalog(_entry(id="a"), _entry(id="b")))
r.acknowledge_entry(session, "a")
assert r.all_entries_acknowledged(session) is False
r.acknowledge_entry(session, "b")
assert r.all_entries_acknowledged(session) is True
def test_no_acknowledge_all_function_exists():
"""Deliberate: there must be no shortcut to acknowledge every entry at
once. See the comment in catalog_review.py above SignDecision."""
names = [n for n in dir(r) if "acknowledge" in n.lower()]
assert "acknowledge_all" not in names
assert "acknowledge_all_entries" not in names
# --------------------------------------------------------------------------- #
# can_sign: TOCTOU blob pinning + acknowledge gating combined
# --------------------------------------------------------------------------- #
def test_can_sign_false_when_not_all_acknowledged():
session = r.start_review("sha1", _catalog(), _catalog(_entry()))
decision = r.can_sign(session, "sha1")
assert decision.ok is False
assert "acknowledged" in decision.reason
def test_can_sign_true_when_acknowledged_and_blob_matches():
session = r.start_review("sha1", _catalog(), _catalog(_entry()))
r.acknowledge_entry(session, "filesystem")
decision = r.can_sign(session, "sha1")
assert decision.ok is True
assert decision.reason is None
def test_can_sign_refuses_on_blob_mismatch_even_if_acknowledged():
"""The core TOCTOU fix: acknowledging everything is not enough if the
bytes on the remote changed underneath the review."""
session = r.start_review("sha1", _catalog(), _catalog(_entry()))
r.acknowledge_entry(session, "filesystem")
decision = r.can_sign(session, "sha2-a-new-commit-landed")
assert decision.ok is False
assert "changed" in decision.reason.lower() or "mismatch" in decision.reason.lower()
def test_can_sign_blob_mismatch_takes_priority_message():
session = r.start_review("sha1", _catalog(), _catalog(_entry()))
decision = r.can_sign(session, "sha2")
assert decision.ok is False
assert "blob" in decision.reason.lower() or "changed" in decision.reason.lower()
# --------------------------------------------------------------------------- #
# catalog_signing_message: domain separation must match bcc_core exactly
# --------------------------------------------------------------------------- #
def test_signing_message_uses_bcc_core_domain_prefix():
raw = b'{"schema":1}'
msg = r.catalog_signing_message(raw)
assert msg == c._CATALOG_SIG_DOMAIN + raw
assert msg.startswith(b"bcc-catalog-v1|")
def test_sign_then_verify_round_trips_with_bcc_core():
"""End-to-end: a signature produced by the Console's sign_catalog_bytes
must verify with bcc_core.verify_catalog_signature -- proves the two
modules can never drift on the domain-separation prefix."""
seed, pubkey = r.generate_keypair()
raw = b'{"schema":1,"version":2,"servers":[]}'
sig = r.sign_catalog_bytes(raw, seed)
assert c.verify_catalog_signature(raw, sig, [pubkey]) is True
def test_sign_tampered_bytes_fails_verify():
seed, pubkey = r.generate_keypair()
raw = b'{"schema":1,"version":2,"servers":[]}'
sig = r.sign_catalog_bytes(raw, seed)
tampered = raw[:-1] + b"0"
assert c.verify_catalog_signature(tampered, sig, [pubkey]) is False
# --------------------------------------------------------------------------- #
# Key encryption at rest
# --------------------------------------------------------------------------- #
def test_encrypt_decrypt_round_trip():
seed, _pub = r.generate_keypair()
blob = r.encrypt_private_key(seed, "correct horse battery staple")
decrypted = r.decrypt_private_key(blob, "correct horse battery staple")
assert decrypted == seed
def test_decrypt_wrong_passphrase_raises():
seed, _pub = r.generate_keypair()
blob = r.encrypt_private_key(seed, "right passphrase")
with pytest.raises(ValueError):
r.decrypt_private_key(blob, "wrong passphrase")
def test_decrypt_corrupted_blob_raises():
seed, _pub = r.generate_keypair()
blob = r.encrypt_private_key(seed, "pass")
corrupted = blob[:-1] + bytes([blob[-1] ^ 0xFF])
with pytest.raises(ValueError):
r.decrypt_private_key(corrupted, "pass")
def test_encrypt_private_key_rejects_wrong_length_seed():
with pytest.raises(ValueError):
r.encrypt_private_key(b"too-short", "pass")
def test_encrypt_private_key_rejects_empty_passphrase():
seed, _pub = r.generate_keypair()
with pytest.raises(ValueError):
r.encrypt_private_key(seed, "")
def test_encrypted_blob_never_contains_seed_plaintext():
seed, _pub = r.generate_keypair()
blob = r.encrypt_private_key(seed, "some passphrase")
assert seed not in blob
def test_generate_keypair_produces_valid_ed25519_pair():
seed, pubkey = r.generate_keypair()
assert len(seed) == 32
assert len(pubkey) == 32
raw = b"test payload"
sig = r.sign_catalog_bytes(raw, seed)
assert c.verify_catalog_signature(raw, sig, [pubkey]) is True
# --------------------------------------------------------------------------- #
# Registry lookup / near-neighbour edit distance
# --------------------------------------------------------------------------- #
def test_edit_distance_identical():
assert r.edit_distance("abc", "abc") == 0
def test_edit_distance_one_substitution():
assert r.edit_distance("firecrawl-mcp", "f1recrawl-mcp") == 1
def test_near_neighbor_ids_finds_close_match():
others = ["firecrawl-mcp", "unrelated-server", "totally-different"]
neighbors = r.near_neighbor_ids("firecrawl-mcp2", others, max_distance=2)
assert "firecrawl-mcp" in neighbors
def test_near_neighbor_ids_excludes_self():
others = ["filesystem", "other"]
assert "filesystem" not in r.near_neighbor_ids("filesystem", others)
def test_near_neighbor_ids_excludes_far_matches():
others = ["completely-unrelated-name"]
assert r.near_neighbor_ids("filesystem", others, max_distance=2) == []
def test_extract_package_refs_npm():
entry = _entry(config={"command": "npx", "args": ["-y", "@scope/pkg@1.2.3"]})
refs = r.extract_package_refs(entry)
assert len(refs) == 1
assert refs[0].ecosystem == "npm"
assert refs[0].name == "@scope/pkg"
assert refs[0].version == "1.2.3"
def test_extract_package_refs_uvx():
entry = _entry(config={"command": "uvx", "args": ["some-pypi-pkg==1.0.0"]})
refs = r.extract_package_refs(entry)
assert len(refs) == 1
assert refs[0].ecosystem == "pypi"
assert refs[0].name == "some-pypi-pkg"
assert refs[0].version == "1.0.0"
def test_extract_package_refs_link_only_entry_returns_empty():
entry = {"id": "slack", "setup": "link-only", "docs_url": "https://example.com"}
assert r.extract_package_refs(entry) == []
def test_lookup_registry_info_fails_soft_on_none():
ref = r.PackageRef("x", "npm", "somepkg", "1.0.0")
info = r.lookup_registry_info(ref, lambda _ref: None, [])
assert info.available is False
def test_lookup_registry_info_fails_soft_on_exception():
def boom(_ref):
raise RuntimeError("network down")
ref = r.PackageRef("x", "npm", "somepkg", "1.0.0")
info = r.lookup_registry_info(ref, boom, [])
assert info.available is False
def test_lookup_registry_info_populates_fields_when_available():
ref = r.PackageRef("x", "npm", "somepkg", "1.0.0")
def fetcher(_ref):
return {
"publisher": "hello_sideguide",
"age_days": 30,
"last_release": "2026-01-01",
"downloads": 500,
}
info = r.lookup_registry_info(ref, fetcher, [])
assert info.available is True
assert info.publisher == "hello_sideguide"
assert info.downloads == 500
def test_lookup_registry_info_includes_near_neighbors():
ref = r.PackageRef("x", "npm", "firecrawl-mcp2", "1.0.0")
info = r.lookup_registry_info(ref, lambda _ref: None, ["firecrawl-mcp"])
assert "firecrawl-mcp" in info.near_neighbor_ids
# --------------------------------------------------------------------------- #
# contains_non_ascii
# --------------------------------------------------------------------------- #
def test_contains_non_ascii_true():
assert r.contains_non_ascii("pаckage") is True
def test_contains_non_ascii_false():
assert r.contains_non_ascii("package") is False