From f0d0ab7a083d32605eacac5988e729aa4b871bd6 Mon Sep 17 00:00:00 2001 From: BCC Agent Date: Sun, 12 Jul 2026 17:57:00 -0400 Subject: [PATCH] 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