better-claude-config/catalog_console.py

"""
catalog_console.py -- Catalog Console: maintainer-only review + signing tool
for data/catalog.json (issue #62).

    MAINTAINER-ONLY. Run from a source checkout. NEVER shipped to users and
    NEVER included in the release bundle -- see bcc.spec (Analysis only ever
    starts from bcc.py) and tests/test_packaging.py, which asserts this file
    and catalog_review.py are absent from the packaged bundle.

Flow: Load -> Review -> Sign.

  1. Load    -- pick a source: an open Gitea PR touching data/catalog.json,
                or the current tip of `main`. The Console fetches the exact
                git blob (via a local clone's git plumbing) and PINS its
                blob SHA for the rest of this review pass.
  2. Review  -- a semantic diff (catalog_review.diff_catalogs), one card per
                changed entry, with risk annotations
                (catalog_review.entry_risk_findings). A registry lookup for
                each entry's npm/PyPI package kicks off automatically, one
                worker thread per entry, the moment the cards are built --
                it is the one check a reviewer can't do by eye, so it must
                never depend on a click. It fails soft (a dead registry
                shows "unavailable", never blocks review or Sign) and a
                per-card "Re-check" button covers manual retries. Every
                changed entry must be individually acknowledged (its
                checkbox ticked) before Sign unlocks. There is no
                "acknowledge all" -- see catalog_review.py.
  3. Sign    -- re-fetches the current blob SHA and refuses to sign unless
                it still matches the pinned SHA from step 1 (TOCTOU fix:
                catalog_review.can_sign). On success, writes
                data/catalog.json + data/catalog.json.sig and commits BOTH
                in a single commit, then pushes -- so main is never red
                between a catalog merge and its signature.

The signature must be the artefact of an actual review, not a step that
follows one. Signing IS the approval act.
"""

from __future__ import annotations

import argparse
import contextlib
import getpass
import html
import json
import re
import subprocess
import sys
import urllib.error
import urllib.request
from dataclasses import dataclass
from pathlib import Path

import bcc_core as core
import catalog_review as review

# --------------------------------------------------------------------------- #
# Constants
# --------------------------------------------------------------------------- #

GITEA_HOST = "git.avezzano.io"
GITEA_API_BASE = f"https://{GITEA_HOST}/api/v1"
REPO_OWNER = "the_og"
REPO_NAME = "better-claude-config"
CATALOG_PATH = "data/catalog.json"
SIG_PATH = "data/catalog.json.sig"

# Outside the repo, per issue #62 ("never committed, never plaintext"). A
# maintainer-only tool, so a dotfile under $HOME is an acceptable fallback
# when the OS keychain isn't available -- the blob stored there is always
# passphrase-encrypted (see catalog_review.encrypt_private_key), never raw.
KEY_STORAGE_DIR = Path.home() / ".bcc-catalog-console"
KEY_STORAGE_FILE = KEY_STORAGE_DIR / "signing_key.enc"

HTTP_TIMEOUT = 6.0


# --------------------------------------------------------------------------- #
# Key storage: OS keychain if available, else a passphrase-encrypted file
# outside the repo. Never plaintext, never an env var, never committed.
# --------------------------------------------------------------------------- #


def _keyring_module():
    """Best-effort import of the optional `keyring` package. Returns None if
    it isn't installed -- this tool must work without it, falling back to
    the encrypted-file path. `keyring` is deliberately NOT added to
    requirements-dev.txt: this is a maintainer-only tool excluded from the
    shipped app, so it doesn't need to justify a new runtime dependency for
    every user the way bcc.py's dependencies do."""
    try:
        import keyring

        return keyring
    except ImportError:
        return None


_KEYRING_SERVICE = "bcc-catalog-console"
_KEYRING_USERNAME = "signing-key"


def store_encrypted_key(blob: bytes) -> str:
    """Persist an already-encrypted key blob (see
    catalog_review.encrypt_private_key). Prefers the OS keychain; falls back
    to a file under KEY_STORAGE_DIR (outside the repo) with restrictive
    permissions. Returns a human-readable description of where it went."""
    keyring = _keyring_module()
    if keyring is not None:
        try:
            keyring.set_password(_KEYRING_SERVICE, _KEYRING_USERNAME, blob.hex())
            return "OS keychain (via the `keyring` package)"
        except Exception:
            pass  # fall through to the file-based path
    KEY_STORAGE_DIR.mkdir(parents=True, exist_ok=True)
    KEY_STORAGE_FILE.write_bytes(blob)
    with contextlib.suppress(OSError):  # best-effort on platforms without POSIX perm bits
        KEY_STORAGE_FILE.chmod(0o600)
    return f"encrypted file at {KEY_STORAGE_FILE}"


def load_encrypted_key() -> bytes:
    """Load the encrypted key blob from wherever store_encrypted_key() put
    it. Raises FileNotFoundError if no key has been generated yet."""
    keyring = _keyring_module()
    if keyring is not None:
        try:
            hex_blob = keyring.get_password(_KEYRING_SERVICE, _KEYRING_USERNAME)
            if hex_blob:
                return bytes.fromhex(hex_blob)
        except Exception:
            pass
    if not KEY_STORAGE_FILE.exists():
        raise FileNotFoundError(
            f"No signing key found (checked the OS keychain and {KEY_STORAGE_FILE}). "
            "Run `python catalog_console.py keygen` first."
        )
    return KEY_STORAGE_FILE.read_bytes()


def unlock_signing_key(passphrase: str) -> bytes:
    """Load + decrypt the signing key seed. Raises ValueError on a wrong
    passphrase, FileNotFoundError if no key exists yet."""
    blob = load_encrypted_key()
    return review.decrypt_private_key(blob, passphrase)


# --------------------------------------------------------------------------- #
# git plumbing against a local clone. The clone's `origin` remote is assumed
# to already carry credentials (the "tokened remote" every other BCC
# maintainer script relies on) -- this module never handles a token itself.
# --------------------------------------------------------------------------- #


class GitError(RuntimeError):
    pass


def _git(repo_dir: Path, *args: str, capture_bytes: bool = False):
    cmd = ["git", "-C", str(repo_dir), *args]
    result = subprocess.run(cmd, capture_output=True, check=False)
    if result.returncode != 0:
        stderr = result.stderr.decode("utf-8", "replace")
        raise GitError(f"git {' '.join(args)} failed: {stderr}")
    return result.stdout if capture_bytes else result.stdout.decode("utf-8", "replace")


def fetch_ref(repo_dir: Path, ref: str) -> str:
    """Fetch `ref` from origin and return the resulting commit SHA."""
    _git(repo_dir, "fetch", "origin", ref)
    return _git(repo_dir, "rev-parse", "FETCH_HEAD").strip()


def blob_sha_at(repo_dir: Path, commit: str, path: str) -> str:
    """The git blob SHA of `path` as it exists at `commit`. This is what
    gets pinned at review-start and re-checked immediately before signing
    (catalog_review.can_sign) -- the TOCTOU fix."""
    return _git(repo_dir, "rev-parse", f"{commit}:{path}").strip()


def blob_bytes(repo_dir: Path, blob_sha: str) -> bytes:
    return _git(repo_dir, "cat-file", "blob", blob_sha, capture_bytes=True)


def read_catalog_at_commit(repo_dir: Path, commit: str) -> tuple[bytes, str]:
    """Return (raw_bytes, blob_sha) for data/catalog.json at `commit`."""
    sha = blob_sha_at(repo_dir, commit, CATALOG_PATH)
    return blob_bytes(repo_dir, sha), sha


def commit_and_push_signed_catalog(
    repo_dir: Path, raw_bytes: bytes, signature: bytes, *, branch: str = "main"
) -> str:
    """Write data/catalog.json + data/catalog.json.sig and commit BOTH in a
    single commit, then push to `branch`. Returns the new commit SHA.

    This is deliberate: if signing happened in a commit AFTER the catalog
    merge, main would be red (payload present, signature missing) between
    every catalog merge and its signing commit. Routine red-main trains
    exactly the alarm fatigue this whole design exists to prevent. Emitting
    one commit with both files means main is never in that state.
    """
    _git(repo_dir, "checkout", branch)
    _git(repo_dir, "pull", "--ff-only", "origin", branch)

    (repo_dir / CATALOG_PATH).write_bytes(raw_bytes)
    (repo_dir / SIG_PATH).write_bytes(signature)

    _git(repo_dir, "add", CATALOG_PATH, SIG_PATH)
    _git(
        repo_dir,
        "commit",
        "-m",
        "chore: sign data/catalog.json (Catalog Console, #62)\n\n"
        "Payload and detached Ed25519 signature land together so main is "
        "never red between a catalog merge and its signature.",
    )
    _git(repo_dir, "push", "origin", branch)
    return _git(repo_dir, "rev-parse", "HEAD").strip()


# --------------------------------------------------------------------------- #
# Gitea REST API: list open PRs touching data/catalog.json
# --------------------------------------------------------------------------- #


def _gitea_get(path: str, token: str | None = None) -> object:
    url = f"{GITEA_API_BASE}{path}"
    req = urllib.request.Request(url)
    if token:
        req.add_header("Authorization", f"token {token}")
    with urllib.request.urlopen(req, timeout=HTTP_TIMEOUT) as resp:
        return json.loads(resp.read().decode("utf-8"))


@dataclass(frozen=True)
class CatalogPR:
    number: int
    title: str
    head_ref: str  # refs/pull/<n>/head


def list_open_catalog_prs(token: str | None = None) -> list[CatalogPR]:
    """Open PRs against REPO_OWNER/REPO_NAME whose diff touches
    data/catalog.json. Fails soft: on any network error, returns [] rather
    than raising into the GUI (Load still offers the `main` source)."""
    try:
        prs = _gitea_get(f"/repos/{REPO_OWNER}/{REPO_NAME}/pulls?state=open", token)
    except (urllib.error.URLError, TimeoutError, ValueError):
        return []

    matches: list[CatalogPR] = []
    for pr in prs or []:
        number = pr.get("number")
        if not isinstance(number, int):
            continue
        if _pr_touches_catalog(number, token):
            matches.append(
                CatalogPR(
                    number=number,
                    title=str(pr.get("title", f"PR #{number}")),
                    head_ref=f"refs/pull/{number}/head",
                )
            )
    return matches


def _pr_touches_catalog(pr_number: int, token: str | None) -> bool:
    url = f"https://{GITEA_HOST}/{REPO_OWNER}/{REPO_NAME}/pulls/{pr_number}.diff"
    req = urllib.request.Request(url)
    if token:
        req.add_header("Authorization", f"token {token}")
    try:
        with urllib.request.urlopen(req, timeout=HTTP_TIMEOUT) as resp:
            diff_text = resp.read().decode("utf-8", "replace")
    except (urllib.error.URLError, TimeoutError):
        return False
    return CATALOG_PATH in diff_text


def token_from_git_remote(repo_dir: Path) -> str | None:
    """Best-effort extraction of a PAT embedded in `origin`'s URL
    (https://<token>@host/...), matching the "tokened remote" every other
    BCC maintainer flow already relies on. Returns None if there isn't one
    (public read-only API calls still work, just rate-limited)."""
    try:
        url = _git(repo_dir, "remote", "get-url", "origin").strip()
    except GitError:
        return None
    match = re.match(r"https://([^@/]+)@", url)
    if not match:
        return None
    token = match.group(1)
    # `user:token` form -- keep only the token half if present.
    return token.split(":", 1)[-1]


# --------------------------------------------------------------------------- #
# Registry lookup fetchers (npm / PyPI). Kept out of catalog_review.py so the
# pure module never makes a network call itself -- these are injected as the
# `Fetcher` callable review.lookup_registry_info() expects.
# --------------------------------------------------------------------------- #


def fetch_npm_info(ref: review.PackageRef) -> dict | None:
    url = f"https://registry.npmjs.org/{ref.name}"
    try:
        req = urllib.request.Request(url, headers={"Accept": "application/json"})
        with urllib.request.urlopen(req, timeout=HTTP_TIMEOUT) as resp:
            data = json.loads(resp.read().decode("utf-8"))
    except (urllib.error.URLError, TimeoutError, ValueError):
        return None

    time_info = data.get("time") or {}
    created = time_info.get("created")
    modified = time_info.get("modified")
    age_days = _iso_age_days(created)
    maintainers = data.get("maintainers") or []
    publisher = maintainers[0].get("name") if maintainers else None

    downloads = None
    try:
        dl_url = f"https://api.npmjs.org/downloads/point/last-week/{ref.name}"
        with urllib.request.urlopen(dl_url, timeout=HTTP_TIMEOUT) as resp:
            downloads = json.loads(resp.read().decode("utf-8")).get("downloads")
    except (urllib.error.URLError, TimeoutError, ValueError):
        pass  # fail soft -- downloads are a nice-to-have, not required

    return {
        "publisher": publisher,
        "age_days": age_days,
        "last_release": modified,
        "downloads": downloads,
    }


def fetch_pypi_info(ref: review.PackageRef) -> dict | None:
    url = f"https://pypi.org/pypi/{ref.name}/json"
    try:
        with urllib.request.urlopen(url, timeout=HTTP_TIMEOUT) as resp:
            data = json.loads(resp.read().decode("utf-8"))
    except (urllib.error.URLError, TimeoutError, ValueError):
        return None

    info = data.get("info") or {}
    releases = data.get("releases") or {}
    last_release = None
    earliest_upload = None
    for files in releases.values():
        for f in files:
            uploaded = f.get("upload_time_iso_8601")
            if not uploaded:
                continue
            if last_release is None or uploaded > last_release:
                last_release = uploaded
            if earliest_upload is None or uploaded < earliest_upload:
                earliest_upload = uploaded

    return {
        "publisher": info.get("author") or info.get("maintainer"),
        "age_days": _iso_age_days(earliest_upload),
        "last_release": last_release,
        "downloads": None,  # PyPI JSON API doesn't include download counts
    }


def _iso_age_days(iso_timestamp: str | None) -> int | None:
    if not iso_timestamp:
        return None
    import datetime as _dt

    try:
        parsed = _dt.datetime.fromisoformat(iso_timestamp.replace("Z", "+00:00"))
        now = _dt.datetime.now(_dt.timezone.utc)
        return max((now - parsed).days, 0)
    except ValueError:
        return None


def registry_fetcher(ref: review.PackageRef) -> dict | None:
    """The Fetcher passed to review.lookup_registry_info(). Never raises --
    both fetch_npm_info/fetch_pypi_info already fail soft, and
    lookup_registry_info() wraps this in a try/except regardless."""
    if ref.ecosystem == "npm":
        return fetch_npm_info(ref)
    if ref.ecosystem == "pypi":
        return fetch_pypi_info(ref)
    return None


# --------------------------------------------------------------------------- #
# GUI (PySide6). Everything above this line has no Qt dependency and is
# exercised by tests/test_catalog_review.py; everything below is a thin
# shell that calls into it.
# --------------------------------------------------------------------------- #

from PySide6.QtCore import Qt, QThread, Signal  # noqa: E402
from PySide6.QtWidgets import (  # noqa: E402
    QApplication,
    QCheckBox,
    QDialog,
    QDialogButtonBox,
    QFormLayout,
    QGroupBox,
    QHBoxLayout,
    QLabel,
    QLineEdit,
    QListWidget,
    QListWidgetItem,
    QMainWindow,
    QMessageBox,
    QPushButton,
    QScrollArea,
    QVBoxLayout,
    QWidget,
)


def plain_label(text: object) -> QLabel:
    """A QLabel guaranteed to render `text` as plain text, never HTML.

    Qt's QLabel auto-interprets HTML by default (Qt.AutoText), which means
    an attacker-controlled description/notes/URL/package-name string
    containing `<b>` or `<img onerror=...>` would render as markup instead
    of visible text -- exactly the kind of thing that could hide a homoglyph
    swap or make a risk warning easy to miss. Every catalog-derived string
    shown by this Console MUST go through this helper (or otherwise set
    Qt.PlainText explicitly) rather than a bare QLabel(...).
    """
    label = QLabel(html.escape(str(text)))
    label.setTextFormat(Qt.PlainText)
    label.setWordWrap(True)
    return label


_SEVERITY_PREFIX = {"blocking": "✖ BLOCKING", "warning": "⚠ WARNING", "info": "ℹ INFO"}


class RegistryLookupWorker(QThread):
    """Off-UI-thread registry lookups, mirroring bcc.py's ConnTester/
    SpawnTester pattern. Never blocks the review UI on a slow/dead network."""

    done = Signal(object)  # list[review.RegistryInfo]

    def __init__(self, refs: list[review.PackageRef], all_entry_ids: list[str]):
        super().__init__()
        self._refs = refs
        self._all_entry_ids = all_entry_ids

    def run(self):
        results = [
            review.lookup_registry_info(ref, registry_fetcher, self._all_entry_ids)
            for ref in self._refs
        ]
        self.done.emit(results)


class EntryCard(QWidget):
    """One changed catalog entry: the diff, risk findings, and the
    acknowledge checkbox that gates Sign. `command`/`args` are rendered
    visually dominant (bold-weight, larger, first) since they're the fields
    that execute.
    """

    acknowledged_changed = Signal(str, bool)

    def __init__(self, change: review.EntryChange, all_entry_ids: list[str]):
        super().__init__()
        self.change = change
        self._all_entry_ids = all_entry_ids
        self._worker: RegistryLookupWorker | None = None

        outline = QVBoxLayout(self)
        box = QGroupBox(f"[{change.status.upper()}] {change.entry_id}")
        outline.addWidget(box)
        layout = QVBoxLayout(box)

        entry = change.new or change.old or {}
        config = entry.get("config") or {}

        cmd_label = plain_label(f"command: {config.get('command', '(none)')}")
        cmd_label.setStyleSheet("font-weight: bold; font-size: 13pt;")
        layout.addWidget(cmd_label)

        args_label = plain_label(f"args: {config.get('args', [])}")
        args_label.setStyleSheet("font-weight: bold;")
        layout.addWidget(args_label)

        for fc in change.field_changes:
            if fc.field in ("config.command", "config.args"):
                continue  # already shown dominant, above
            layout.addWidget(plain_label(f"{fc.field}: {fc.old!r} -> {fc.new!r}"))

        findings = review.entry_risk_findings(change)
        for finding in findings:
            prefix = _SEVERITY_PREFIX.get(finding.severity, finding.severity.upper())
            flabel = plain_label(f"{prefix}: {finding.message}")
            if finding.severity == "blocking":
                flabel.setStyleSheet("color: #c62828; font-weight: bold;")
            elif finding.severity == "warning":
                flabel.setStyleSheet("color: #ef6c00;")
            else:
                flabel.setStyleSheet("color: #1565c0;")
            layout.addWidget(flabel)

        self.registry_label = plain_label("Registry lookup: loading...")
        layout.addWidget(self.registry_label)
        recheck_btn = QPushButton("Re-check")
        recheck_btn.clicked.connect(self._run_registry_lookup)
        layout.addWidget(recheck_btn)

        self.blocking = any(f.severity == "blocking" for f in findings)
        self.checkbox = QCheckBox(
            "I have reviewed this entry, including command/args and the risk"
            " annotations above, and approve it."
        )
        if self.blocking:
            self.checkbox.setEnabled(False)
            self.checkbox.setToolTip(
                "This entry has a BLOCKING finding and cannot be acknowledged "
                "until the underlying change is fixed (edit the PR, don't sign around it)."
            )
        self.checkbox.toggled.connect(
            lambda checked: self.acknowledged_changed.emit(change.entry_id, checked)
        )
        layout.addWidget(self.checkbox)

        # Registry lookup is the one check a reviewer can't do by eye -- it's
        # what catches a typosquatted/hijacked package (it already caught
        # firecrawl-mcp in the seed data). It must run automatically as soon
        # as the card exists, not wait on a click a tired maintainer might
        # skip at 11pm. Off the GUI thread (RegistryLookupWorker is a
        # QThread) and fails soft: a dead/slow registry can never gate
        # review or signing, it just leaves this entry's lookup showing
        # "unavailable". The "Re-check" button above stays for retrying a
        # failed/unavailable lookup by hand.
        self._run_registry_lookup()

    def _run_registry_lookup(self):
        entry = self.change.new or {}
        refs = review.extract_package_refs(entry)
        if not refs:
            self.registry_label.setText("Registry lookup: no npm/PyPI package in this entry.")
            return
        self.registry_label.setText("Registry lookup: loading...")
        self._worker = RegistryLookupWorker(refs, self._all_entry_ids)
        self._worker.done.connect(self._on_registry_result)
        self._worker.start()

    def _on_registry_result(self, results: list[review.RegistryInfo]):
        lines = []
        for info in results:
            if not info.available:
                lines.append(f"{info.ref.name}: unavailable (network/registry unreachable)")
                continue
            neighbor_note = (
                f" | NEAR-NEIGHBOUR of: {', '.join(info.near_neighbor_ids)}"
                if info.near_neighbor_ids
                else ""
            )
            lines.append(
                f"{info.ref.name}: publisher={info.publisher!r} age_days={info.age_days} "
                f"last_release={info.last_release} downloads={info.downloads}{neighbor_note}"
            )
        text = "Registry lookup:\n" + "\n".join(lines)
        self.registry_label.setText(html.escape(text))
        self.registry_label.setTextFormat(Qt.PlainText)


class PassphraseDialog(QDialog):
    def __init__(self, prompt: str, parent=None):
        super().__init__(parent)
        self.setWindowTitle("Signing key passphrase")
        layout = QFormLayout(self)
        self.edit = QLineEdit()
        self.edit.setEchoMode(QLineEdit.EchoMode.Password)
        layout.addRow(prompt, self.edit)
        buttons = QDialogButtonBox(
            QDialogButtonBox.StandardButton.Ok | QDialogButtonBox.StandardButton.Cancel
        )
        buttons.accepted.connect(self.accept)
        buttons.rejected.connect(self.reject)
        layout.addRow(buttons)

    def passphrase(self) -> str:
        return self.edit.text()


class ReviewWindow(QMainWindow):
    def __init__(self, repo_dir: Path):
        super().__init__()
        self.repo_dir = repo_dir
        self.session: review.ReviewSession | None = None
        self.cards: dict[str, EntryCard] = {}

        self.setWindowTitle("BCC Catalog Console -- maintainer-only, never shipped")
        central = QWidget()
        self.setCentralWidget(central)
        root = QVBoxLayout(central)

        top = QHBoxLayout()
        self.source_list = QListWidget()
        self.source_list.addItem(QListWidgetItem("main (current tip)"))
        top.addWidget(self.source_list, 1)

        side = QVBoxLayout()
        load_btn = QPushButton("Load selected source")
        load_btn.clicked.connect(self._on_load)
        side.addWidget(load_btn)
        refresh_prs_btn = QPushButton("Refresh open PR list")
        refresh_prs_btn.clicked.connect(self._refresh_pr_list)
        side.addWidget(refresh_prs_btn)
        side.addStretch(1)
        top.addLayout(side)
        root.addLayout(top)

        self.scroll = QScrollArea()
        self.scroll.setWidgetResizable(True)
        self.card_container = QWidget()
        self.card_layout = QVBoxLayout(self.card_container)
        self.scroll.setWidget(self.card_container)
        root.addWidget(self.scroll, 1)

        self.status_label = plain_label("Load a source to begin review.")
        root.addWidget(self.status_label)

        self.sign_btn = QPushButton("Sign")
        self.sign_btn.setEnabled(False)
        self.sign_btn.clicked.connect(self._on_sign)
        root.addWidget(self.sign_btn)

        self._token = token_from_git_remote(self.repo_dir)
        self._prs: list[CatalogPR] = []
        self._refresh_pr_list()

    def _refresh_pr_list(self):
        self._prs = list_open_catalog_prs(self._token)
        while self.source_list.count() > 1:
            self.source_list.takeItem(1)
        for pr in self._prs:
            self.source_list.addItem(QListWidgetItem(f"PR #{pr.number}: {pr.title}"))

    def _on_load(self):
        row = self.source_list.currentRow()
        try:
            if row <= 0:
                commit = fetch_ref(self.repo_dir, "main")
                old_commit = None  # main vs itself has no "old" -- nothing to diff without a base
            else:
                pr = self._prs[row - 1]
                commit = fetch_ref(self.repo_dir, pr.head_ref)
                old_commit = fetch_ref(self.repo_dir, "main")

            new_raw, new_blob_sha = read_catalog_at_commit(self.repo_dir, commit)
            new_catalog = core.load_catalog(new_raw)

            if old_commit:
                old_raw, _old_sha = read_catalog_at_commit(self.repo_dir, old_commit)
                old_catalog = core.load_catalog(old_raw)
            else:
                old_catalog = new_catalog

        except (GitError, ValueError) as e:
            QMessageBox.critical(self, "Load failed", html.escape(str(e)))
            return

        self._new_raw = new_raw
        self.session = review.start_review(new_blob_sha, old_catalog, new_catalog)
        self._render_cards()

    def _render_cards(self):
        while self.card_layout.count():
            item = self.card_layout.takeAt(0)
            if item.widget():
                item.widget().deleteLater()
        self.cards.clear()

        assert self.session is not None
        all_ids = sorted(
            {e.get("id") for e in (self.session.new_catalog.get("servers") or []) if e.get("id")}
        )
        for change in self.session.changes:
            card = EntryCard(change, all_ids)
            card.acknowledged_changed.connect(self._on_acknowledge_changed)
            self.cards[change.entry_id] = card
            self.card_layout.addWidget(card)
        self.card_layout.addStretch(1)
        self._update_status()

    def _on_acknowledge_changed(self, entry_id: str, checked: bool):
        assert self.session is not None
        if checked:
            review.acknowledge_entry(self.session, entry_id)
        else:
            review.unacknowledge_entry(self.session, entry_id)
        self._update_status()

    def _update_status(self):
        assert self.session is not None
        all_ack = review.all_entries_acknowledged(self.session)
        self.sign_btn.setEnabled(all_ack)
        pending = len(self.session.changes) - len(self.session.acknowledged)
        self.status_label.setText(
            f"{len(self.session.changes)} changed entries, {pending} not yet acknowledged."
        )

    def _on_sign(self):
        assert self.session is not None
        try:
            current_sha = blob_sha_at(self.repo_dir, fetch_ref(self.repo_dir, "main"), CATALOG_PATH)
        except GitError as e:
            QMessageBox.critical(self, "Sign failed", html.escape(str(e)))
            return

        decision = review.can_sign(self.session, current_sha)
        if not decision.ok:
            QMessageBox.warning(self, "Cannot sign", html.escape(decision.reason or ""))
            if decision.reason and "changed" in decision.reason.lower():
                self._on_load()  # force a re-review against the new bytes
            return

        dialog = PassphraseDialog("Enter signing key passphrase:", self)
        if dialog.exec() != QDialog.DialogCode.Accepted:
            return
        try:
            seed = unlock_signing_key(dialog.passphrase())
        except (FileNotFoundError, ValueError) as e:
            QMessageBox.critical(self, "Sign failed", html.escape(str(e)))
            return

        signature = review.sign_catalog_bytes(self._new_raw, seed)
        try:
            new_commit = commit_and_push_signed_catalog(self.repo_dir, self._new_raw, signature)
        except GitError as e:
            QMessageBox.critical(self, "Commit/push failed", html.escape(str(e)))
            return

        QMessageBox.information(self, "Signed", f"Signed and pushed as commit {new_commit[:12]}.")
        self.sign_btn.setEnabled(False)


# --------------------------------------------------------------------------- #
# CLI
# --------------------------------------------------------------------------- #


def cmd_keygen(_args: argparse.Namespace) -> int:
    seed, pubkey = review.generate_keypair()
    passphrase = getpass.getpass("Choose a passphrase to encrypt the new signing key: ")
    confirm = getpass.getpass("Confirm passphrase: ")
    if passphrase != confirm:
        print("error: passphrases did not match", file=sys.stderr)
        return 1
    if not passphrase:
        print("error: a non-empty passphrase is required", file=sys.stderr)
        return 1

    blob = review.encrypt_private_key(seed, passphrase)
    where = store_encrypted_key(blob)
    pubkey_b64 = __import__("base64").b64encode(pubkey).decode("ascii")

    print(f"Private key encrypted and stored in: {where}")
    print()
    print("Public key (base64, paste into bcc_core.CATALOG_PUBKEYS):")
    print(f"    {pubkey_b64}")
    print()
    print(
        "Also add it as the Gitea repo secret RELEASE_SIGNING_KEY (base64 of the "
        "32-byte private seed) used by release.yml -- get that value with:"
    )
    print("    python catalog_console.py show-seed-b64   # careful: prints the raw key")
    return 0


def cmd_show_seed_b64(_args: argparse.Namespace) -> int:
    passphrase = getpass.getpass("Signing key passphrase: ")
    try:
        seed = unlock_signing_key(passphrase)
    except (FileNotFoundError, ValueError) as e:
        print(f"error: {e}", file=sys.stderr)
        return 1
    import base64

    print(base64.b64encode(seed).decode("ascii"))
    return 0


def cmd_gui(args: argparse.Namespace) -> int:
    repo_dir = Path(args.repo).resolve()
    if not (repo_dir / CATALOG_PATH).exists():
        print(
            f"error: {repo_dir} doesn't look like a BCC checkout (no {CATALOG_PATH})",
            file=sys.stderr,
        )
        return 1

    app = QApplication(sys.argv)
    app.setApplicationName("BCC Catalog Console")
    win = ReviewWindow(repo_dir)
    win.resize(900, 700)
    win.show()
    return app.exec()


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description=__doc__)
    sub = parser.add_subparsers(dest="command")

    p_gui = sub.add_parser("gui", help="launch the review/sign GUI (default)")
    p_gui.add_argument("--repo", default=".", help="path to a BCC git checkout (default: cwd)")
    p_gui.set_defaults(func=cmd_gui)

    p_keygen = sub.add_parser("keygen", help="generate a new Ed25519 signing keypair")
    p_keygen.set_defaults(func=cmd_keygen)

    p_seed = sub.add_parser(
        "show-seed-b64", help="print the base64 private seed (for the RELEASE_SIGNING_KEY secret)"
    )
    p_seed.set_defaults(func=cmd_show_seed_b64)

    return parser


_SUBCOMMANDS = ("gui", "keygen", "show-seed-b64", "-h", "--help")


def main(argv: list[str] | None = None) -> int:
    argv = sys.argv[1:] if argv is None else list(argv)
    # `python catalog_console.py` with no subcommand (or with GUI-only flags
    # like --repo) launches the GUI -- "gui" is the default action.
    if not argv or argv[0] not in _SUBCOMMANDS:
        argv = ["gui", *argv]
    parser = build_parser()
    args = parser.parse_args(argv)
    return args.func(args)


if __name__ == "__main__":
    raise SystemExit(main())