From f0d0ab7a083d32605eacac5988e729aa4b871bd6 Mon Sep 17 00:00:00 2001 From: BCC Agent Date: Sun, 12 Jul 2026 17:57:00 -0400 Subject: [PATCH 1/2] feat: Catalog Console -- maintainer-only review + signing tool (#62) Adds a separate PySide6 tool (catalog_console.py) that reviews proposed changes to data/catalog.json and signs the approved result. Never shipped to users, never in the release bundle -- maintainer runs it from source. Pure, GUI-free logic lives in a new catalog_review.py (kept out of both the GUI and bcc_core.py to avoid merge conflicts on the latter): - diff_catalogs(old, new) -> list[EntryChange]: semantic (per-entry) diff, not a text diff, with per-field before/after values. - Six independent risk predicates, each unit-tested: non-empty env_required value, command outside bcc_core.CATALOG_ALLOWED_COMMANDS (imported, not redefined), non-ASCII code points in id/command/args (rendered with escapes -- homoglyph/RTL-override defence), unpinned npm/docker package references, URL domain changes (lookalike-domain swap defence), brand-new entries flagged for extra scrutiny. - ReviewSession + can_sign(): the Sign button stays disabled until every changed entry is individually acknowledged -- no "acknowledge all" shortcut exists, and a comment in the code says never to add one. - TOCTOU fix (adversarial review on #62): the git blob SHA of data/catalog.json is pinned when review begins; can_sign() refuses to sign if the current blob differs, forcing a re-review. The Console re-fetches the blob SHA immediately before signing and enforces this. - catalog_signing_message() imports bcc_core's domain-separation prefix (_CATALOG_SIG_DOMAIN) rather than retyping it, so the Console's signatures and bcc_core.verify_catalog_signature can't drift apart -- proven by a round-trip test (sign here, verify via bcc_core). - encrypt_private_key/decrypt_private_key: the signing key is never stored plaintext (scrypt + AES-256-GCM at rest, OS keychain via the optional keyring package if available, else an encrypted file under $HOME outside the repo). - Registry lookup (lookup_registry_info + injected Fetcher): the network call is kept out of this module for offline testability; catalog_console.py supplies npm/PyPI HTTP fetchers. Fails soft -- network down means "unavailable", never a block on review. near_neighbor_ids() flags edit-distance <=2 typosquat candidates against existing catalog ids. catalog_console.py wires the above into a Qt GUI: Load (open PRs touching data/catalog.json via the Gitea REST API, or main) -> Review (one EntryCard per changed entry, command/args rendered visually dominant, risk findings colour-coded, per-card registry-lookup button running off the UI thread like bcc.py's ConnTester/SpawnTester) -> Sign (re-checks the pinned blob SHA, prompts for the key passphrase, writes data/catalog.json + data/catalog.json.sig and commits+pushes BOTH in a single commit -- so main is never red between a catalog merge and its signature). Every attacker-controlled string renders through a plain_label() helper that both escapes HTML and forces Qt.PlainText, so a script/image payload in a description/notes/URL can't render as markup. Also provides keygen (generates + stores an encrypted keypair, prints the base64 public key) and show-seed-b64 (prints the base64 private seed for the RELEASE_SIGNING_KEY CI secret) CLI subcommands. Excluded from the release bundle: bcc.spec's Analysis() only ever starts from bcc.py, and tests/test_catalog_console_packaging.py asserts neither new file is named anywhere in bcc.spec and that bcc.py never imports either module. Tests: 67 new (62 in test_catalog_review.py, 5 in test_catalog_console_packaging.py) covering diff_catalogs, every risk predicate individually, the acknowledge-gating + TOCTOU can_sign() logic, the sign/verify round-trip against bcc_core, key encryption (including wrong-passphrase and corrupted-blob rejection), edit-distance/near-neighbour matching, and registry-lookup fail-soft behaviour. Full suite: 321 passed, 1 pre-existing unrelated skip. ruff check and ruff format --check both clean. catalog_console.py (Qt/GUI) could not be executed in the sandbox this was developed in (no system EGL/GL libraries available for PySide6) -- it was syntax-checked (py_compile) and lint/format-checked but not smoke-tested; see PR body for what AJ should verify. Closes #62 --- catalog_console.py | 822 ++++++++++++++++++++++++ catalog_review.py | 727 +++++++++++++++++++++ tests/test_catalog_console_packaging.py | 66 ++ tests/test_catalog_review.py | 555 ++++++++++++++++ 4 files changed, 2170 insertions(+) create mode 100644 catalog_console.py create mode 100644 catalog_review.py create mode 100644 tests/test_catalog_console_packaging.py create mode 100644 tests/test_catalog_review.py diff --git a/catalog_console.py b/catalog_console.py new file mode 100644 index 0000000..0d6f007 --- /dev/null +++ b/catalog_console.py @@ -0,0 +1,822 @@ +""" +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) and a live registry + lookup. 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//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://@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 `` or `` 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: not checked yet.") + layout.addWidget(self.registry_label) + check_btn = QPushButton("Check registry") + check_btn.clicked.connect(self._run_registry_lookup) + layout.addWidget(check_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) + + 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: checking...") + 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()) diff --git a/catalog_review.py b/catalog_review.py new file mode 100644 index 0000000..6edf044 --- /dev/null +++ b/catalog_review.py @@ -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 ` -- 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)) diff --git a/tests/test_catalog_console_packaging.py b/tests/test_catalog_console_packaging.py new file mode 100644 index 0000000..4112b85 --- /dev/null +++ b/tests/test_catalog_console_packaging.py @@ -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 diff --git a/tests/test_catalog_review.py b/tests/test_catalog_review.py new file mode 100644 index 0000000..adee2b6 --- /dev/null +++ b/tests/test_catalog_review.py @@ -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 From d6fc6845c4106640784c6ad23e6efd8f29b20edd Mon Sep 17 00:00:00 2001 From: BCC Agent Date: Sun, 12 Jul 2026 18:04:28 -0400 Subject: [PATCH 2/2] fix(catalog-console): run registry lookup automatically, not on click (#62) Review feedback: making the registry lookup an on-demand 'Check registry' button was a deviation from the design intent, not a style choice. The registry lookup is the one check a human reviewer genuinely cannot do by eye -- it's what caught firecrawl-mcp's unrelated npm publisher in the seed data. Gating it behind a button makes it optional, and an optional check is the one a tired maintainer skips at 11pm -- exactly the failure mode this tool exists to defend against. The friction belongs on approval, never on information. EntryCard now kicks off its registry lookup automatically at construction time (i.e. as soon as _render_cards() builds the cards for a loaded review), one RegistryLookupWorker (QThread) per changed entry, all starting concurrently as the cards are built. Nothing about the lookup's pure logic changed -- catalog_review.lookup_registry_info was already fail-soft (RegistryInfo(available=False) on any fetch problem, never an exception) and can_sign() never depended on registry state, so a dead registry still cannot gate review or signing. Renamed the button 'Check registry' -> 'Re-check' and kept it wired to the same _run_registry_lookup(), for manually retrying a failed/unavailable lookup. Label copy now reads loading... while a lookup is in flight (was 'not checked yet.' / 'checking...'), matching the loading -> result | unavailable per-entry states. No change to acknowledge-gating or the TOCTOU blob-SHA pin in catalog_review.py. ruff check/format clean; full suite still 321 passed, 1 pre-existing unrelated skip -- catalog_review.py (the tested pure-logic module) is untouched, only catalog_console.py's GUI wiring moved from button-triggered to auto-triggered. --- catalog_console.py | 35 ++++++++++++++++++++++++++--------- 1 file changed, 26 insertions(+), 9 deletions(-) diff --git a/catalog_console.py b/catalog_console.py index 0d6f007..0438379 100644 --- a/catalog_console.py +++ b/catalog_console.py @@ -15,10 +15,16 @@ Flow: Load -> Review -> Sign. 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) and a live registry - lookup. Every changed entry must be individually - acknowledged (its checkbox ticked) before Sign unlocks. - There is no "acknowledge all" -- see catalog_review.py. + (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 @@ -497,11 +503,11 @@ class EntryCard(QWidget): flabel.setStyleSheet("color: #1565c0;") layout.addWidget(flabel) - self.registry_label = plain_label("Registry lookup: not checked yet.") + self.registry_label = plain_label("Registry lookup: loading...") layout.addWidget(self.registry_label) - check_btn = QPushButton("Check registry") - check_btn.clicked.connect(self._run_registry_lookup) - layout.addWidget(check_btn) + 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( @@ -519,13 +525,24 @@ class EntryCard(QWidget): ) 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: checking...") + 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()