fix(core): validate config.env, enforce version pinning, fix CI trust anchor and resolve_catalog guards (#68)

Fixes findings 2, 3, 4, 6, 7 from the issue #68 adversarial review.

- Finding 2: config.env was type-checked only. Add CATALOG_DENIED_ENV_KEYS
  (case-insensitive) for interpreter/loader-override keys (NODE_OPTIONS,
  PYTHONPATH, LD_PRELOAD, ...), apply the ASCII check and the existing
  secret-value check to env keys/values, and require env values to be
  empty or a single <PLACEHOLDER> token.

- Finding 3: version pinning was only checked by catalog_review.py (which
  never runs on the signing path per finding 1). Move enforcement into
  _validate_catalog_config: npm/uvx specs must carry @version or ==version
  (scoped names handled), docker images must have an explicit non-latest
  tag. Only the first plausible package-spec token is checked, so flags,
  <PLACEHOLDER>s, and docker subcommands/flags don't trip it. All 19 real
  catalog entries still validate clean.

- Finding 4: the CI catalog-signature gate imported bcc_core from the PR
  branch and trusted whatever CATALOG_PUBKEYS said there, so a PR changing
  both catalog.json and CATALOG_PUBKEYS (with a matching signature) went
  green. ci.yml now hardcodes the expected base64 pubkey and asserts
  bcc_core.CATALOG_PUBKEYS matches it before verifying the signature.
  NOTE: the maintainer is planning to rotate this key -- update
  EXPECTED_CATALOG_PUBKEY_B64 in ci.yml as its own reviewed change when
  that happens, never bundled with a catalog content change.

- Finding 6: resolve_catalog's anti-rollback/anti-freeze guards sat behind
  `if best_version >= 0`, so the first verified candidate was accepted
  unconditionally and the anti-freeze anchor drifted with each accepted
  candidate instead of staying fixed. The cap is now measured against the
  bundled catalog's version specifically (the trust anchor baked into the
  binary), regardless of evaluation order; bundled wins version ties; and
  a new pure `floor` parameter lets a future caller pass a persisted
  accepted-version floor.

- Finding 7: catalog id is now constrained to ^[a-z0-9][a-z0-9._-]{0,63}$.

Tests: fixed _minimal_catalog to use a pinned package (was enshrining
finding 3), rewrote the env-passthrough test to prove the validation
boundary instead of asserting env passes through unchecked, and
reordered test_resolve_catalog_rejects_absurd_version_jump so it
actually exercises the first-candidate path. Added positive/negative
tests for every new rule. Manually verified each new check by commenting
it out and confirming the guarding test goes red, then restoring it.

bcc_core.py

@@ -2164,6 +2164,30 @@ def restart_claude_desktop() -> RestartResult:
 # is rejected by validate_catalog() regardless of how plausible it looks.
 CATALOG_ALLOWED_COMMANDS = frozenset({"npx", "uvx", "docker", "node", "python", "python3"})
 
+# Env var keys a catalog entry's config.env must never set. Every one of
+# these is a loader/interpreter override that lets a value walk straight
+# past CATALOG_ALLOWED_COMMANDS and the -e/--eval/-c deny-rule below: e.g.
+# NODE_OPTIONS="--require /tmp/x.js" turns an allowlisted `npx` entry into
+# arbitrary code execution without ever touching config.args, which is the
+# only field the allowlist/deny-rules/ASCII/secret checks used to cover.
+# Matched case-insensitively -- env keys are case-sensitive on POSIX, but a
+# `node_options` lookalike is exactly the kind of thing this must catch.
+CATALOG_DENIED_ENV_KEYS = frozenset(
+    {
+        "NODE_OPTIONS",
+        "PYTHONSTARTUP",
+        "PYTHONPATH",
+        "PYTHONHOME",
+        "LD_PRELOAD",
+        "LD_LIBRARY_PATH",
+        "DYLD_INSERT_LIBRARIES",
+        "DYLD_LIBRARY_PATH",
+        "BROWSER",
+        "PATH",
+        "NODE_REPL_EXTERNAL_MODULE",
+    }
+)
+
 # Ed25519 public keys allowed to sign a catalog, raw 32-byte form. A LIST
 # (not a single key) so keys can be rotated without bricking installs that
 # still trust an older key: verify_catalog_signature() accepts a match
@@ -2189,6 +2213,36 @@ _CATALOG_SECRET_ARG_RE = re.compile(r"(?i)--api[-_]?key=|--token=|--password=")
 # <PLACEHOLDER>-style tokens the GUI must have the user fill in before Save.
 _PLACEHOLDER_RE = re.compile(r"<[^<>\s]+>")
 
+# A catalog entry's id becomes an mcpServers JSON key AND is interpolated
+# into Qt.AutoText widgets (status bar, QMessageBox) -- an id like
+# "<b>Verified</b>" renders as markup there. Not RCE, but UI spoofing, so
+# ids are constrained to a plain lowercase slug.
+_CATALOG_ID_RE = re.compile(r"^[a-z0-9][a-z0-9._-]{0,63}$")
+
+# Docker flags that consume the next arg as a value (so that value must not
+# be mistaken for the image reference when locating it in config.args).
+_DOCKER_VALUE_FLAGS = frozenset(
+    {
+        "-e",
+        "--env",
+        "-v",
+        "--volume",
+        "-p",
+        "--publish",
+        "--name",
+        "-w",
+        "--workdir",
+        "-u",
+        "--user",
+        "--entrypoint",
+        "--network",
+        "--platform",
+        "--add-host",
+        "-l",
+        "--label",
+    }
+)
+
 # How many versions a single accepted catalog jump may leap in one go. Bounds
 # a "freeze" attack: a compromised/leaked signing key claiming an absurd
 # future version would otherwise permanently outrank every legitimate
@@ -2263,6 +2317,121 @@ def _docker_arg_violations(tag: str, args: list[str]) -> list[str]:
     return problems
 
 
+def _catalog_package_spec_version(spec: str) -> str | None:
+    """
+    Extract the version pin from an npm-style package spec, or None if the
+    spec carries no pin.
+
+    Handles unscoped "name@version" and scoped "@scope/name@version" --
+    scoped names have a leading "@" that is NOT the version separator, so a
+    naive split on the first/only "@" misparses "@scope/pkg" (no version)
+    as pinned to "scope/pkg". Splitting from the right side instead is safe
+    for both forms because a package name may contain "@" only as the
+    scope's leading character.
+    """
+    if spec.startswith("@"):
+        rest = spec[1:]
+        if "@" not in rest:
+            return None
+        _, _, version = rest.rpartition("@")
+        return version or None
+    if "@" not in spec:
+        return None
+    _, _, version = spec.rpartition("@")
+    return version or None
+
+
+def _catalog_package_spec_pinned(spec: str) -> bool:
+    """
+    True if `spec` carries an exact version pin. Covers npm's "name@version"
+    / "@scope/name@version" and uv's documented PyPI pin forms
+    "name@version" and "name==version".
+    """
+    if "==" in spec:
+        _, _, version = spec.partition("==")
+        return bool(version)
+    return bool(_catalog_package_spec_version(spec))
+
+
+def _first_catalog_package_spec(args: list[str]) -> str | None:
+    """
+    The first arg that could plausibly BE a package spec: skip flags
+    (leading "-") and <PLACEHOLDER> tokens (which can't be validated and
+    are filled in by the user later, never shipped by the catalog as the
+    package name itself). Everything after the first hit is ignored --
+    trailing flags, paths, and placeholders are not package specs.
+    """
+    for a in args:
+        if a.startswith("-"):
+            continue
+        if _PLACEHOLDER_RE.fullmatch(a):
+            continue
+        return a
+    return None
+
+
+def _catalog_pin_violations(tag: str, command: str, args: list[str]) -> list[str]:
+    """
+    Version-pinning enforcement (finding #3): a catalog PR can otherwise
+    ship `npx -y @scope/pkg` or `docker run img:latest` and the *next*
+    resolve of that package/image is whatever the registry serves that day
+    -- outside review, outside the signature's meaning. This is the only
+    place that enforces pinning at runtime; catalog_review.py's
+    risk_unpinned_package() is a maintainer-facing hint, not a gate.
+    """
+    if command in ("npx", "uvx"):
+        spec = _first_catalog_package_spec(args)
+        if spec is None:
+            return [f"{tag}: config.args must include a package spec to pin (e.g. name@1.2.3)."]
+        if not _catalog_package_spec_pinned(spec):
+            return [
+                f"{tag}: config.args package {spec!r} is not version-pinned; use "
+                "name@version, @scope/name@version, or name==version."
+            ]
+        return []
+
+    if command == "docker":
+        image = _docker_image_ref(args)
+        if image is None:
+            return [f"{tag}: config.args docker command has no image reference to pin."]
+        _, sep, image_tag = image.rpartition(":")
+        if not sep or "/" in image_tag:
+            return [
+                f"{tag}: config.args docker image {image!r} has no explicit tag; "
+                "pin an exact version (not 'latest', not untagged)."
+            ]
+        if image_tag == "latest":
+            return [
+                f"{tag}: config.args docker image {image!r} uses the 'latest' tag, "
+                "which is not allowed; pin an exact version."
+            ]
+        return []
+
+    return []
+
+
+def _docker_image_ref(args: list[str]) -> str | None:
+    """
+    Locate the image reference in a `docker run ...` args list: skip the
+    "run" subcommand and any flags, including ones that consume the next
+    token as a value (-e, -v, --name, ...) so that value isn't mistaken for
+    the image. The first remaining positional token is the image.
+    """
+    i = 0
+    if i < len(args) and args[i] == "run":
+        i += 1
+    while i < len(args):
+        a = args[i]
+        if a.startswith("-"):
+            if a in _DOCKER_VALUE_FLAGS and "=" not in a:
+                i += 2
+            else:
+                i += 1
+            continue
+        return a
+    return None
+
+
 def _validate_catalog_config(tag: str, config) -> list[str]:
     """Validate the `config` block of a basic-tier catalog entry."""
     if not isinstance(config, dict):
@@ -2283,10 +2452,11 @@ def _validate_catalog_config(tag: str, config) -> list[str]:
             f"({', '.join(sorted(CATALOG_ALLOWED_COMMANDS))})."
         )
 
-    args = config.get("args")
-    if not isinstance(args, list) or not all(isinstance(a, str) for a in args):
+    raw_args = config.get("args")
+    args_ok = isinstance(raw_args, list) and all(isinstance(a, str) for a in raw_args)
+    args = raw_args if args_ok else []
+    if not args_ok:
         problems.append(f"{tag}: config.args must be a list of strings.")
-        args = []
 
     for a in args:
         if not a.isascii():
@@ -2305,11 +2475,55 @@ def _validate_catalog_config(tag: str, config) -> list[str]:
     if command == "docker":
         problems.extend(_docker_arg_violations(tag, args))
 
+    # Version pinning (finding #3) -- only meaningful once command/args are
+    # actually well-formed; a malformed args list already got its own
+    # problem above and has nothing left to pin-check.
+    if args_ok and command in ("npx", "uvx", "docker"):
+        problems.extend(_catalog_pin_violations(tag, command, args))
+
     env = config.get("env")
-    if env is not None and (
-        not isinstance(env, dict) or any(not isinstance(v, str) for v in env.values())
-    ):
-        problems.append(f"{tag}: config.env must be an object of string values.")
+    if env is not None:
+        env_ok = isinstance(env, dict) and all(
+            isinstance(k, str) and isinstance(v, str) for k, v in env.items()
+        )
+        if not env_ok:
+            problems.append(f"{tag}: config.env must be an object of string values.")
+        else:
+            # config.env (finding #2): unlike args, env was previously
+            # type-checked ONLY -- no allowlist, no deny-rule, no ASCII
+            # check, no secret check. That made it the single easiest way
+            # to smuggle a payload past every other guard in this
+            # function: an allowlisted `command: npx` plus
+            # NODE_OPTIONS=--require /tmp/x.js in env walks straight past
+            # the command allowlist AND the -e/--eval/-c deny-rule above,
+            # because neither of those ever looks at env.
+            for key, value in env.items():
+                if not key.isascii():
+                    problems.append(
+                        f"{tag}: config.env key {key!r} must be ASCII "
+                        "(non-ASCII code points rejected)."
+                    )
+                if key.upper() in CATALOG_DENIED_ENV_KEYS:
+                    problems.append(
+                        f"{tag}: config.env key {key!r} is on the catalog deny-list "
+                        "(interpreter/loader override) and is not allowed."
+                    )
+                if not value.isascii():
+                    problems.append(
+                        f"{tag}: config.env value for {key!r} must be ASCII "
+                        "(non-ASCII code points rejected)."
+                    )
+                if _is_secret_value(value):
+                    problems.append(
+                        f"{tag}: config.env[{key!r}] looks like a real secret value; "
+                        "catalog entries must never ship secret values."
+                    )
+                if value != "" and not _PLACEHOLDER_RE.fullmatch(value):
+                    problems.append(
+                        f"{tag}: config.env[{key!r}] must be an empty string or a "
+                        "single <PLACEHOLDER> token -- the catalog declares which env "
+                        "vars a server needs, it never supplies their values."
+                    )
 
     return problems
 
@@ -2329,6 +2543,12 @@ def _validate_catalog_entry(idx: int, entry, seen_ids: set[str]) -> list[str]:
         tag = f"servers[{idx}] ({entry_id!r})"
         if not entry_id.isascii():
             problems.append(f"{tag}: 'id' must be ASCII (non-ASCII code points rejected).")
+        elif not _CATALOG_ID_RE.match(entry_id):
+            problems.append(
+                f"{tag}: 'id' must match ^[a-z0-9][a-z0-9._-]{{0,63}}$ "
+                "(it becomes an mcpServers JSON key and is interpolated into "
+                "Qt.AutoText widgets)."
+            )
         if entry_id in seen_ids:
             problems.append(f"{tag}: duplicate id.")
         seen_ids.add(entry_id)
@@ -2442,15 +2662,32 @@ def verify_catalog_signature(raw: bytes, sig: bytes, pubkeys: list[bytes]) -> bo
     return False
 
 
+def _verify_catalog_candidate(candidate: tuple[bytes, bytes] | None) -> tuple[dict | None, int]:
+    """Verify+load+validate one (raw, sig) candidate. Returns (None, -1) on any failure."""
+    if not candidate:
+        return None, -1
+    raw, sig = candidate
+    if not verify_catalog_signature(raw, sig, CATALOG_PUBKEYS):
+        return None, -1
+    try:
+        data = load_catalog(raw)
+    except (ValueError, TypeError):
+        return None, -1
+    if validate_catalog(data):
+        return None, -1
+    return data, catalog_version(data)
+
+
 def resolve_catalog(
     bundled: tuple[bytes, bytes] | None,
     cached: tuple[bytes, bytes] | None,
     remote: tuple[bytes, bytes] | None,
+    floor: int = 0,
 ) -> dict:
     """
-    Pick the highest-version catalog among bundled/cached/remote. Each
-    argument is either None (unavailable) or an (raw_bytes, signature_bytes)
-    pair.
+    Pick the highest-version catalog among bundled/cached/remote. Each of
+    bundled/cached/remote is either None (unavailable) or an (raw_bytes,
+    signature_bytes) pair.
 
     🔴 SECURITY: every candidate — including `bundled`, the copy frozen into
     this binary — is verified against CATALOG_PUBKEYS and re-validated from
@@ -2461,46 +2698,68 @@ def resolve_catalog(
     by virtue of being local. Signing (and checking the signature at
     runtime, every time) closes that.
 
-    Anti-rollback: a candidate's version is never accepted if it's lower
-    than the best verified candidate already found in this same resolution
-    pass — an attacker replaying an old, since-superseded signed catalog
-    can't downgrade you.
+    `floor` is a pure, caller-supplied lower bound (e.g. a persisted
+    "last accepted version" the GUI can load from disk and pass in) — this
+    function does no storage of its own.
 
-    Anti-freeze: a candidate whose version leaps more than
-    _CATALOG_MAX_VERSION_JUMP past the current best is also rejected. A
-    compromised/leaked signing key claiming an absurd future version would
-    otherwise permanently outrank every legitimate release from then on,
-    since the resolver always prefers the highest verified version — this
-    caps how far a single accepted jump can go.
+    Anti-rollback / anti-freeze, and WHY they apply to every candidate
+    including the first one evaluated: the previous version of this
+    function only ran these checks `if best_version >= 0`, i.e. once a
+    candidate had already been accepted in this pass. That let the FIRST
+    verified candidate through unconditionally — a signed catalog claiming
+    version=999999999 sailed straight past both guards if it happened to be
+    evaluated first, and rollback protection reset on every call anyway
+    (nothing persisted across restarts). Now both guards are anchored to
+    something that doesn't depend on iteration order:
+
+    - The anti-freeze cap is measured against the BUNDLED catalog's version
+      (verified independently, once), not against "whatever was accepted
+      so far in this loop." Bundled ships inside the binary, so it's the
+      one candidate that isn't attacker-supplied at resolve time — the
+      natural trust anchor. If bundled itself doesn't verify, `floor` is
+      the anchor instead.
+    - The anti-rollback floor is max(floor, bundled's version), so a
+      caller that persists `floor` across restarts gets real rollback
+      protection; a caller that doesn't still gets "never below bundled."
+
+    On a version TIE, the bundled candidate wins over cached/remote (it
+    previously lost ties to whichever candidate happened to be evaluated
+    last, silently preferring remote over bundled at equal version).
 
     Returns the winning catalog dict, or {} if nothing verified and
     validated.
     """
+    bundled_data, bundled_version = _verify_catalog_candidate(bundled)
+
+    anchor = bundled_version if bundled_version >= 0 else floor
+    min_accepted = max(floor, bundled_version if bundled_version >= 0 else 0)
+
+    candidates = (
+        ("bundled", bundled_data, bundled_version),
+        ("cached", *_verify_catalog_candidate(cached)),
+        ("remote", *_verify_catalog_candidate(remote)),
+    )
+
     best: dict = {}
     best_version = -1
+    best_is_bundled = False
 
-    for candidate in (bundled, cached, remote):
-        if not candidate:
-            continue
-        raw, sig = candidate
-        if not verify_catalog_signature(raw, sig, CATALOG_PUBKEYS):
-            continue
-        try:
-            data = load_catalog(raw)
-        except (ValueError, TypeError):
-            continue
-        if validate_catalog(data):
+    for source, data, version in candidates:
+        if data is None:
             continue
+        if version < min_accepted:
+            continue  # anti-rollback / below the persisted floor
+        if version > anchor + _CATALOG_MAX_VERSION_JUMP:
+            continue  # anti-freeze, capped against the bundled trust anchor
 
-        version = catalog_version(data)
-        if best_version >= 0:
-            if version < best_version:
-                continue  # anti-rollback
-            if version > best_version + _CATALOG_MAX_VERSION_JUMP:
-                continue  # anti-freeze
-
-        best = data
-        best_version = version
+        is_bundled = source == "bundled"
+        better = version > best_version or (
+            version == best_version and is_bundled and not best_is_bundled
+        )
+        if better:
+            best = data
+            best_version = version
+            best_is_bundled = is_bundled
 
     return best