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

feat: Catalog Console -- maintainer-only review + signing tool (#62)
CI / Lint (ruff) (pull_request) Successful in 7s
Details
CI / Tests (py3.12 / windows-latest) (pull_request) Successful in 23s
Details
CI / Tests (py3.12 / ubuntu-latest) (pull_request) Successful in 12s
Details
CI / Tests (py3.13 / ubuntu-latest) (pull_request) Successful in 10s
Details
CI / Tests (py3.10 / ubuntu-latest) (pull_request) Successful in 10s
Details

Browse Source 
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
This commit is contained in:
BCC Agent
2026-07-12 17:57:00 -04:00
parent 3841106630
commit f0d0ab7a08
4 changed files with 2170 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
catalog_review.py
+727
View File
@@ -0,0 +1,727 @@
"""
catalog_review.py -- pure, GUI-free review/diff/risk/crypto logic for the
Catalog Console (issue #62).
This module is deliberately Qt-free and network-free so every function in it
is unit-testable offline, exactly like bcc_core.py. catalog_console.py (the
PySide6 GUI) is a thin shell over these functions -- it owns Qt widgets,
subprocess/git calls, and HTTP registry lookups; this module owns judgment.
Nothing here is reimplemented from bcc_core: the command allowlist and the
signature domain-separation prefix are imported, not retyped, so the two
modules cannot silently drift apart (see bcc_core.validate_catalog /
bcc_core.verify_catalog_signature and the project's "the check drifted on a
new surface" recurring-bug lesson).
"""
from __future__ import annotations
import os
import re
from collections.abc import Callable
from dataclasses import dataclass, field
from urllib.parse import urlsplit
from bcc_core import _CATALOG_SIG_DOMAIN as CATALOG_SIG_DOMAIN
from bcc_core import CATALOG_ALLOWED_COMMANDS
# --------------------------------------------------------------------------- #
# Semantic diff
# --------------------------------------------------------------------------- #
# Top-level scalar/simple fields compared directly (not drilled into).
_DIFF_FIELDS = (
"display",
"description",
"category",
"official",
"setup",
"homepage",
"docs_url",
"source",
"notes",
"stars",
"last_release",
"env_required",
)
@dataclass(frozen=True)
class FieldChange:
"""One field that differs between the old and new version of an entry."""
field: str
old: object
new: object
@dataclass(frozen=True)
class EntryChange:
"""One catalog entry's change: added, removed, or changed.
`old`/`new` are the raw entry dicts (or None for added/removed) so risk
predicates and the GUI can inspect anything not captured by
`field_changes` (which only lists fields that actually differ).
"""
entry_id: str
status: str # "added" | "removed" | "changed"
old: dict | None
new: dict | None
field_changes: tuple[FieldChange, ...] = ()
def _config_field_changes(old_cfg: dict | None, new_cfg: dict | None) -> list[FieldChange]:
old_cfg = old_cfg or {}
new_cfg = new_cfg or {}
changes: list[FieldChange] = []
for f in ("command", "args", "env"):
ov, nv = old_cfg.get(f), new_cfg.get(f)
if ov != nv:
changes.append(FieldChange(f"config.{f}", ov, nv))
return changes
def _entry_field_changes(old_entry: dict, new_entry: dict) -> tuple[FieldChange, ...]:
changes: list[FieldChange] = []
for f in _DIFF_FIELDS:
ov, nv = old_entry.get(f), new_entry.get(f)
if ov != nv:
changes.append(FieldChange(f, ov, nv))
changes.extend(_config_field_changes(old_entry.get("config"), new_entry.get("config")))
return tuple(changes)
def diff_catalogs(old: dict | None, new: dict | None) -> list[EntryChange]:
"""Semantic (per-entry) diff between two parsed catalog dicts.
NOT a text diff: entries are matched by `id`, and each changed entry
reports exactly which fields differ (with before/after values), which is
what lets the Console render "command changed from X to Y" instead of a
JSON line diff a reviewer has to mentally reconstruct.
Entries missing/malformed `id` are ignored here -- that is a
validate_catalog() rejection, not a diffing concern, and diffing must not
silently invent a match for two differently-broken entries.
"""
old_servers = {
e["id"]: e
for e in (old or {}).get("servers", []) or []
if isinstance(e, dict) and isinstance(e.get("id"), str) and e.get("id")
}
new_servers = {
e["id"]: e
for e in (new or {}).get("servers", []) or []
if isinstance(e, dict) and isinstance(e.get("id"), str) and e.get("id")
}
changes: list[EntryChange] = []
for entry_id in sorted(set(old_servers) | set(new_servers)):
old_e = old_servers.get(entry_id)
new_e = new_servers.get(entry_id)
if old_e is None:
changes.append(EntryChange(entry_id, "added", None, new_e, ()))
elif new_e is None:
changes.append(EntryChange(entry_id, "removed", old_e, None, ()))
elif old_e != new_e:
fc = _entry_field_changes(old_e, new_e)
if fc:
changes.append(EntryChange(entry_id, "changed", old_e, new_e, fc))
return changes
# --------------------------------------------------------------------------- #
# Risk annotations -- each predicate is pure and independently unit-tested.
# --------------------------------------------------------------------------- #
@dataclass(frozen=True)
class RiskFinding:
severity: str # "blocking" | "warning" | "info"
code: str
message: str
def _escape_non_ascii(s: str) -> str:
"""Render a string with any non-ASCII code point shown as an escape
sequence, so a homoglyph/RTL-override character can't visually pass as
the real thing in the review UI."""
return s.encode("unicode_escape").decode("ascii")
def risk_env_required(change: EntryChange) -> list[RiskFinding]:
"""A non-empty env_required value is blocking: catalog entries must ship
only the *names* of env vars the user fills in, never values."""
entry = change.new or {}
env_required = entry.get("env_required")
findings: list[RiskFinding] = []
if isinstance(env_required, dict):
for k, v in env_required.items():
if v not in (None, ""):
findings.append(
RiskFinding(
"blocking",
"env_required_value",
f"env_required[{k!r}] carries a non-empty value -- catalog "
"entries must never ship secret values, only placeholder names.",
)
)
return findings
def risk_command_allowlist(change: EntryChange) -> list[RiskFinding]:
"""A command outside bcc_core.CATALOG_ALLOWED_COMMANDS is blocking.
Imports the allowlist rather than redefining it."""
entry = change.new or {}
config = entry.get("config") or {}
command = config.get("command")
if isinstance(command, str) and command and command not in CATALOG_ALLOWED_COMMANDS:
return [
RiskFinding(
"blocking",
"command_not_allowed",
f"command {command!r} is not on the catalog allowlist "
f"({', '.join(sorted(CATALOG_ALLOWED_COMMANDS))}).",
)
]
return []
def risk_non_ascii(change: EntryChange) -> list[RiskFinding]:
"""Non-ASCII code points in id/command/args are blocking -- homoglyph /
RTL-override typosquatting can make a malicious package name visually
identical to a legitimate one in a naive diff view."""
entry = change.new or {}
findings: list[RiskFinding] = []
entry_id = entry.get("id")
if isinstance(entry_id, str) and not entry_id.isascii():
findings.append(
RiskFinding(
"blocking",
"non_ascii_id",
f"id contains non-ASCII code points: {_escape_non_ascii(entry_id)!r}",
)
)
config = entry.get("config") or {}
command = config.get("command")
if isinstance(command, str) and not command.isascii():
findings.append(
RiskFinding(
"blocking",
"non_ascii_command",
f"command contains non-ASCII code points: {_escape_non_ascii(command)!r}",
)
)
for a in config.get("args") or []:
if isinstance(a, str) and not a.isascii():
findings.append(
RiskFinding(
"blocking",
"non_ascii_arg",
f"arg contains non-ASCII code points: {_escape_non_ascii(a)!r}",
)
)
return findings
def _npm_candidate_args(command: str | None, args: list[str]) -> list[str]:
if command != "npx":
return []
return [a for a in args if isinstance(a, str) and a and not a.startswith("-")]
def split_npm_spec(spec: str) -> tuple[str, str | None]:
"""Split an npm package spec into (name, version). version is None if
unpinned. Handles scoped (@scope/name@version) and unscoped
(name@version) specs."""
if spec.startswith("@"):
rest = spec[1:]
if "/" not in rest:
return spec, None # malformed scope, can't tell -- treat unpinned
scope, _, remainder = rest.partition("/")
if "@" in remainder:
pkg_name, _, version = remainder.partition("@")
return f"@{scope}/{pkg_name}", (version or None)
return f"@{scope}/{remainder}", None
if "@" in spec:
name, _, version = spec.partition("@")
return name, (version or None)
return spec, None
def is_pinned_npm_spec(spec: str) -> bool:
_name, version = split_npm_spec(spec)
return bool(version)
_DOCKER_VALUE_FLAGS = {
"-e",
"--env",
"-v",
"--volume",
"-p",
"--publish",
"-w",
"--workdir",
"-u",
"--user",
"--name",
"--network",
"--entrypoint",
}
def docker_image_candidates(args: list[str]) -> list[str]:
"""Best-effort extraction of the image reference from a `docker run
[OPTIONS] IMAGE [CMD...]` args list: the first positional token after
any leading `run` and flag(+value) pairs."""
candidates: list[str] = []
i = 0
while i < len(args):
a = args[i]
if a == "run":
i += 1
continue
if isinstance(a, str) and a.startswith("-"):
if "=" not in a and a in _DOCKER_VALUE_FLAGS:
i += 2
continue
i += 1
continue
if isinstance(a, str):
candidates.append(a)
break # first positional token after `run` is the image ref
return candidates
def is_pinned_docker_image(image: str) -> bool:
if "@sha256:" in image:
return True
tag_part = image.rsplit("/", 1)[-1]
if ":" not in tag_part:
return False # no tag => implicit :latest
tag = tag_part.rsplit(":", 1)[-1]
return bool(tag) and tag != "latest"
def risk_unpinned_package(change: EntryChange) -> list[RiskFinding]:
"""Every entry must pin an exact version: an `@scope/pkg` npm arg with
no `@version`, or a docker image with no tag / `:latest`, is blocking.
A later-compromised package must not be able to auto-upgrade into every
user just because the catalog entry never pinned a version."""
entry = change.new or {}
config = entry.get("config") or {}
command = config.get("command")
args = config.get("args") or []
findings: list[RiskFinding] = []
if command == "npx":
for a in _npm_candidate_args(command, args):
if not is_pinned_npm_spec(a):
findings.append(
RiskFinding(
"blocking",
"unpinned_npm_package",
f"npm package arg {a!r} has no pinned @version.",
)
)
elif command == "docker":
for img in docker_image_candidates(args):
if not is_pinned_docker_image(img):
findings.append(
RiskFinding(
"blocking",
"unpinned_docker_image",
f"docker image {img!r} is not pinned to an exact tag "
"(uses :latest or no tag).",
)
)
return findings
_URL_FIELDS = ("homepage", "docs_url", "source")
def _domain(url: str) -> str:
try:
return urlsplit(url).netloc.lower()
except ValueError:
return ""
def risk_url_domain_change(change: EntryChange) -> list[RiskFinding]:
"""Non-https URLs and, more importantly, a *domain change* on any URL
field are surfaced loudly with old-vs-new domains broken out -- the
lookalike-domain-swap defence."""
findings: list[RiskFinding] = []
old_entry = change.old or {}
new_entry = change.new or {}
for f in _URL_FIELDS:
new_url = new_entry.get(f)
if not isinstance(new_url, str) or not new_url:
continue
if not new_url.startswith("https://"):
findings.append(
RiskFinding("warning", "non_https_url", f"{f} is not https://: {new_url!r}")
)
old_url = old_entry.get(f)
if isinstance(old_url, str) and old_url:
old_domain, new_domain = _domain(old_url), _domain(new_url)
if old_domain and new_domain and old_domain != new_domain:
findings.append(
RiskFinding(
"warning",
"domain_changed",
f"{f} domain changed from {old_domain!r} to {new_domain!r} -- "
"verify this isn't a lookalike-domain swap.",
)
)
return findings
def risk_new_entry(change: EntryChange) -> list[RiskFinding]:
"""A brand-new entry is flagged for extra scrutiny -- not blocking on its
own, but it's the category of change the registry lookup exists for."""
if change.status == "added":
return [
RiskFinding(
"info",
"new_entry",
"Brand-new catalog entry -- extra scrutiny: check publisher identity "
"via the registry lookup before signing.",
)
]
return []
_RISK_PREDICATES: tuple[Callable[[EntryChange], list[RiskFinding]], ...] = (
risk_env_required,
risk_command_allowlist,
risk_non_ascii,
risk_unpinned_package,
risk_url_domain_change,
risk_new_entry,
)
def entry_risk_findings(change: EntryChange) -> list[RiskFinding]:
"""Run every risk predicate against one entry change and return the
combined findings (order matches _RISK_PREDICATES)."""
findings: list[RiskFinding] = []
for predicate in _RISK_PREDICATES:
findings.extend(predicate(change))
return findings
def has_blocking_risk(change: EntryChange) -> bool:
return any(f.severity == "blocking" for f in entry_risk_findings(change))
# --------------------------------------------------------------------------- #
# Review session: acknowledge-gating + TOCTOU blob-SHA pinning
# --------------------------------------------------------------------------- #
@dataclass
class ReviewSession:
"""State for one review pass. `pinned_blob_sha` is the git blob SHA of
data/catalog.json as it existed the moment review began -- see
can_sign()."""
pinned_blob_sha: str
old_catalog: dict
new_catalog: dict
changes: list[EntryChange] = field(default_factory=list)
acknowledged: set[str] = field(default_factory=set)
def __post_init__(self) -> None:
if not self.changes:
self.changes = diff_catalogs(self.old_catalog, self.new_catalog)
def start_review(pinned_blob_sha: str, old_catalog: dict, new_catalog: dict) -> ReviewSession:
return ReviewSession(
pinned_blob_sha=pinned_blob_sha, old_catalog=old_catalog, new_catalog=new_catalog
)
def acknowledge_entry(session: ReviewSession, entry_id: str) -> None:
ids = {c.entry_id for c in session.changes}
if entry_id not in ids:
raise ValueError(f"{entry_id!r} is not part of this review session's diff.")
session.acknowledged.add(entry_id)
def unacknowledge_entry(session: ReviewSession, entry_id: str) -> None:
session.acknowledged.discard(entry_id)
def all_entries_acknowledged(session: ReviewSession) -> bool:
return {c.entry_id for c in session.changes} <= session.acknowledged
# NOTE for future editors: do NOT add an "acknowledge all" shortcut here, now
# or ever. The friction of individually acknowledging every changed entry is
# the entire point of this tool (issue #62) -- a shortcut would let a tired
# reviewer rubber-stamp a diff exactly like the "merge PR, run script, push"
# reflex this Console exists to replace. If this comment is the only thing
# stopping you, that is the point: it is stopping you on purpose.
@dataclass(frozen=True)
class SignDecision:
ok: bool
reason: str | None = None
def can_sign(session: ReviewSession, current_blob_sha: str) -> SignDecision:
"""Whether the Sign button may fire right now.
Two independent gates, both required:
1. TOCTOU: `current_blob_sha` (fetched fresh, immediately before signing)
must match the blob SHA pinned when review began. If the bytes on the
remote changed since -- a new commit pushed to the same PR, a
force-push, another PR merged in between -- signing is refused and a
re-review is forced. This is what makes "signing is the approval act"
true rather than aspirational: the signature is bound to the exact
reviewed bytes, not to "whatever the file happens to be now".
2. Every changed entry in the diff must be individually acknowledged.
"""
if current_blob_sha != session.pinned_blob_sha:
return SignDecision(
False,
"The reviewed bytes changed since this review began (blob SHA "
"mismatch) -- re-review required before signing.",
)
if not all_entries_acknowledged(session):
pending = sorted({c.entry_id for c in session.changes} - session.acknowledged)
return SignDecision(
False, f"Not every changed entry has been acknowledged yet: {', '.join(pending)}"
)
return SignDecision(True, None)
def catalog_signing_message(raw_bytes: bytes) -> bytes:
"""The exact bytes that get signed: bcc_core's domain-separation prefix
(imported, never retyped) + the raw catalog bytes. Using this function
guarantees the Console's signature and bcc_core.verify_catalog_signature
can never drift apart on the prefix."""
return CATALOG_SIG_DOMAIN + raw_bytes
# --------------------------------------------------------------------------- #
# Key management: passphrase-encrypted-at-rest Ed25519 seed
#
# The private key is NEVER stored plaintext, never an env var, never
# committed. encrypt_private_key/decrypt_private_key are pure and offline
# (scrypt KDF + AES-256-GCM via `cryptography`, already a project
# dependency); catalog_console.py decides WHERE the resulting blob lives
# (OS keychain if available, else a file outside the repo).
# --------------------------------------------------------------------------- #
_KDF_SALT_LEN = 16
_KDF_N = 2**15 # scrypt cost parameter, tuned for a one-off interactive unlock
_KDF_R = 8
_KDF_P = 1
_NONCE_LEN = 12
_AAD = b"bcc-catalog-console-key-v1"
def _derive_key(passphrase: str, salt: bytes) -> bytes:
from cryptography.hazmat.primitives.kdf.scrypt import Scrypt
kdf = Scrypt(salt=salt, length=32, n=_KDF_N, r=_KDF_R, p=_KDF_P)
return kdf.derive(passphrase.encode("utf-8"))
def generate_keypair() -> tuple[bytes, bytes]:
"""Generate a new Ed25519 keypair. Returns (seed_32_bytes, pubkey_32_bytes)."""
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat
private_key = Ed25519PrivateKey.generate()
seed = private_key.private_bytes_raw()
pubkey = private_key.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw)
return seed, pubkey
def encrypt_private_key(seed: bytes, passphrase: str) -> bytes:
"""Encrypt a 32-byte Ed25519 seed at rest with a passphrase. Returns a
self-contained blob: salt || nonce || ciphertext+tag."""
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
if len(seed) != 32:
raise ValueError(f"expected a 32-byte raw Ed25519 seed, got {len(seed)} bytes")
if not passphrase:
raise ValueError("a non-empty passphrase is required")
salt = os.urandom(_KDF_SALT_LEN)
key = _derive_key(passphrase, salt)
nonce = os.urandom(_NONCE_LEN)
ciphertext = AESGCM(key).encrypt(nonce, seed, _AAD)
return salt + nonce + ciphertext
def decrypt_private_key(blob: bytes, passphrase: str) -> bytes:
"""Decrypt a blob produced by encrypt_private_key. Raises ValueError on a
wrong passphrase or corrupt blob -- never silently returns garbage."""
from cryptography.exceptions import InvalidTag
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
if len(blob) < _KDF_SALT_LEN + _NONCE_LEN:
raise ValueError("key blob is too short to be valid")
salt = blob[:_KDF_SALT_LEN]
nonce = blob[_KDF_SALT_LEN : _KDF_SALT_LEN + _NONCE_LEN]
ciphertext = blob[_KDF_SALT_LEN + _NONCE_LEN :]
key = _derive_key(passphrase, salt)
try:
return AESGCM(key).decrypt(nonce, ciphertext, _AAD)
except InvalidTag as e:
raise ValueError("wrong passphrase or corrupted key file") from e
def sign_catalog_bytes(raw: bytes, seed: bytes) -> bytes:
"""Sign `raw` catalog bytes with a 32-byte Ed25519 seed, using the exact
domain-separated message bcc_core.verify_catalog_signature expects."""
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
if len(seed) != 32:
raise ValueError(f"expected a 32-byte raw Ed25519 seed, got {len(seed)} bytes")
private_key = Ed25519PrivateKey.from_private_bytes(seed)
return private_key.sign(catalog_signing_message(raw))
# --------------------------------------------------------------------------- #
# Registry lookup -- the check a human genuinely can't do.
#
# The network call itself is injected (a `Fetcher` callable) so this stays
# testable offline; catalog_console.py supplies the real npm/PyPI HTTP
# fetcher. Fails soft everywhere: a fetcher returning None/raising just
# yields RegistryInfo(available=False), never an exception into the caller
# and never a block on review.
# --------------------------------------------------------------------------- #
@dataclass(frozen=True)
class PackageRef:
entry_id: str
ecosystem: str # "npm" | "pypi"
name: str
version: str | None
def split_pypi_spec(spec: str) -> tuple[str, str | None]:
for sep in ("==", "@"):
if sep in spec:
name, _, version = spec.partition(sep)
return name, (version or None)
return spec, None
def extract_package_refs(entry: dict) -> list[PackageRef]:
"""Pull out the package(s) a basic-tier entry's args reference, for the
registry lookup. Returns [] for link-only entries or entries whose
command isn't npx/uvx (docker images aren't registry-lookup candidates
in the npm/PyPI sense used here)."""
config = entry.get("config") or {}
command = config.get("command")
args = config.get("args") or []
entry_id = entry.get("id", "") if isinstance(entry.get("id"), str) else ""
refs: list[PackageRef] = []
if command == "npx":
for a in _npm_candidate_args(command, args):
name, version = split_npm_spec(a)
if name:
refs.append(PackageRef(entry_id, "npm", name, version))
elif command == "uvx":
for a in args:
if isinstance(a, str) and a and not a.startswith("-"):
name, version = split_pypi_spec(a)
if name:
refs.append(PackageRef(entry_id, "pypi", name, version))
break # `uvx <pkg>` -- first positional token is the package
return refs
@dataclass(frozen=True)
class RegistryInfo:
ref: PackageRef
available: bool
publisher: str | None = None
age_days: int | None = None
last_release: str | None = None
downloads: int | None = None
near_neighbor_ids: tuple[str, ...] = ()
Fetcher = Callable[[PackageRef], dict | None]
def edit_distance(a: str, b: str) -> int:
"""Levenshtein distance, iterative DP (no recursion depth concerns)."""
if a == b:
return 0
la, lb = len(a), len(b)
if la == 0:
return lb
if lb == 0:
return la
prev = list(range(lb + 1))
for i, ca in enumerate(a, 1):
cur = [i] + [0] * lb
for j, cb in enumerate(b, 1):
cost = 0 if ca == cb else 1
cur[j] = min(prev[j] + 1, cur[j - 1] + 1, prev[j - 1] + cost)
prev = cur
return prev[lb]
def near_neighbor_ids(name: str, other_ids: list[str], max_distance: int = 2) -> list[str]:
"""Catalog ids within `max_distance` edits of `name` (case-insensitive),
excluding an exact match -- the dependency-confusion / typosquat
near-neighbour warning."""
lname = name.lower()
return [
oid
for oid in other_ids
if oid != name and edit_distance(lname, oid.lower()) <= max_distance
]
def lookup_registry_info(
ref: PackageRef, fetcher: Fetcher, all_entry_ids: list[str]
) -> RegistryInfo:
"""Resolve one package against the live registry via the injected
fetcher. Never raises: any fetcher exception or falsy return means
`available=False` ("unavailable"), which the GUI renders plainly rather
than blocking or erroring the review."""
neighbors = tuple(near_neighbor_ids(ref.name, all_entry_ids))
try:
raw = fetcher(ref)
except Exception:
raw = None
if not raw:
return RegistryInfo(ref=ref, available=False, near_neighbor_ids=neighbors)
return RegistryInfo(
ref=ref,
available=True,
publisher=raw.get("publisher"),
age_days=raw.get("age_days"),
last_release=raw.get("last_release"),
downloads=raw.get("downloads"),
near_neighbor_ids=neighbors,
)
_NON_ASCII_RE = re.compile(r"[^\x00-\x7f]")
def contains_non_ascii(s: str) -> bool:
return bool(_NON_ASCII_RE.search(s))