diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index ba3ab86..b2cd6c4 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -95,12 +95,39 @@ jobs: - name: Install dependencies run: pip install cryptography + # ๐Ÿ”ด TRUST ANCHOR โ€” issue #68 finding 4. + # + # This step used to do `import bcc_core as c` FROM THE CHECKED-OUT PR + # BRANCH and verify the catalog against c.CATALOG_PUBKEYS โ€” i.e. it + # trusted the public key shipped in the very diff it was reviewing. A + # PR that changed data/catalog.json AND bcc_core.CATALOG_PUBKEYS (to + # an attacker key, with a matching signature produced by the attacker's + # matching private key) went green, because there was nothing outside + # the PR's own content to check the key against. The gate's whole + # point is catching a friendly-looking PR the maintainer merges + # without really reading it โ€” and that hole made it a two-file diff. + # + # EXPECTED_CATALOG_PUBKEY_B64 below is hardcoded HERE, in the workflow + # file, independent of whatever bcc_core.py says on the PR branch. It + # is intentionally the only line in this step that matters for + # security review: changing it changes what this gate is willing to + # trust. THIS CONSTANT IS A TRUST ANCHOR. A PR that changes this line + # in the same diff as a catalog change is exactly the attack this gate + # exists to prevent โ€” review a change to this line on its own, + # never bundled with a catalog update. + # + # NOTE for the next key rotation: update EXPECTED_CATALOG_PUBKEY_B64 + # below to the new key's base64 form, as its own reviewed change. - name: Verify data/catalog.json.sig + env: + EXPECTED_CATALOG_PUBKEY_B64: "082NOwVB7uURkvfyS3+knJ+40Fk6C9unsF47+2uPKo4=" run: | python - <<'PY' - import pathlib, sys + import base64, os, pathlib, sys import bcc_core as c + expected_pubkey_b64 = os.environ["EXPECTED_CATALOG_PUBKEY_B64"] + raw = pathlib.Path("data/catalog.json").read_bytes() sig_path = pathlib.Path("data/catalog.json.sig") @@ -111,6 +138,26 @@ jobs: if b"\x00" * 32 in c.CATALOG_PUBKEYS: sys.exit("FAIL: CATALOG_PUBKEYS still holds the placeholder key.") + # Trust anchor check FIRST, before verifying anything against + # bcc_core.CATALOG_PUBKEYS: a PR is not allowed to bring its own + # key. CATALOG_PUBKEYS on the checked-out branch must be EXACTLY + # the key(s) this workflow file itself expects -- no more, no + # fewer, no substitutions. + actual_pubkeys_b64 = [base64.b64encode(k).decode() for k in c.CATALOG_PUBKEYS] + if actual_pubkeys_b64 != [expected_pubkey_b64]: + sys.exit( + "FAIL: bcc_core.CATALOG_PUBKEYS on this branch does not match the " + "trust anchor hardcoded in .github/workflows/ci.yml.\n" + f" expected: {[expected_pubkey_b64]}\n" + f" actual: {actual_pubkeys_b64}\n" + "\n" + "This PR is changing (or has changed) the catalog signing key. That " + "change must be reviewed on its own, separately from any catalog " + "content change, and the workflow's EXPECTED_CATALOG_PUBKEY_B64 " + "updated deliberately -- not accepted because it happened to match " + "whatever bcc_core.py says on this branch." + ) + if not c.verify_catalog_signature(raw, sig_path.read_bytes(), c.CATALOG_PUBKEYS): sys.exit( "FAIL: data/catalog.json does NOT match its signature.\n" @@ -124,5 +171,6 @@ jobs: if problems: sys.exit("FAIL: catalog failed validation:\n " + "\n ".join(problems)) - print("OK: catalog signature verifies and the catalog validates clean.") + print("OK: catalog signature verifies, the pubkey matches the CI trust anchor, " + "and the catalog validates clean.") PY diff --git a/bcc_core.py b/bcc_core.py index 699c8cf..c81d5ea 100644 --- a/bcc_core.py +++ b/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=") # -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 +# "Verified" 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 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 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 diff --git a/tests/test_core.py b/tests/test_core.py index 8b58cd2..725358f 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -1690,8 +1690,12 @@ def _minimal_catalog(version: int = 1) -> dict: "official": True, "setup": "basic", "config": { + # Pinned on purpose (issue #68 finding 3): an earlier + # version of this fixture used an unpinned package and + # asserted it validated clean, which enshrined the bug + # instead of catching it. "command": "npx", - "args": ["-y", "widget-mcp"], + "args": ["-y", "widget-mcp@1.0.0"], }, "placeholders": {}, "env_required": {}, @@ -1952,6 +1956,213 @@ def test_validate_catalog_rejects_duplicate_ids(): assert any("duplicate id" in p for p in problems) +# --- validate_catalog: config.env (issue #68 finding 2) -------------------- # +def test_validate_catalog_rejects_each_denied_env_key(): + for key in sorted(c.CATALOG_DENIED_ENV_KEYS): + data = _catalog_with( + {"config": {"command": "npx", "args": ["-y", "widget-mcp@1.0.0"], "env": {key: ""}}} + ) + problems = c.validate_catalog(data) + assert any("deny-list" in p for p in problems), (key, problems) + + +def test_validate_catalog_rejects_denied_env_key_case_insensitively(): + data = _catalog_with( + { + "config": { + "command": "npx", + "args": ["-y", "widget-mcp@1.0.0"], + "env": {"node_options": ""}, + } + } + ) + problems = c.validate_catalog(data) + assert any("deny-list" in p for p in problems) + + +def test_validate_catalog_rejects_nonempty_nonplaceholder_env_value(): + data = _catalog_with( + { + "config": { + "command": "npx", + "args": ["-y", "widget-mcp@1.0.0"], + "env": {"FOO_URL": "https://example.com"}, + } + } + ) + problems = c.validate_catalog(data) + assert any("empty string or a single " in p for p in problems) + + +def test_validate_catalog_accepts_empty_and_placeholder_env_values(): + data = _catalog_with( + { + "config": { + "command": "npx", + "args": ["-y", "widget-mcp@1.0.0"], + "env": {"FOO": "", "BAR_URL": ""}, + } + } + ) + assert c.validate_catalog(data) == [] + + +def test_validate_catalog_rejects_non_ascii_env_key(): + data = _catalog_with( + { + "config": { + "command": "npx", + "args": ["-y", "widget-mcp@1.0.0"], + "env": {"Fร–O": ""}, + } + } + ) + problems = c.validate_catalog(data) + assert any("config.env key" in p and "ASCII" in p for p in problems) + + +def test_validate_catalog_rejects_non_ascii_env_value(): + data = _catalog_with( + { + "config": { + "command": "npx", + "args": ["-y", "widget-mcp@1.0.0"], + "env": {"FOO": ""}, + } + } + ) + problems = c.validate_catalog(data) + assert any("config.env value" in p and "ASCII" in p for p in problems) + + +def test_validate_catalog_rejects_secret_looking_env_value(): + data = _catalog_with( + { + "config": { + "command": "npx", + "args": ["-y", "widget-mcp@1.0.0"], + "env": {"SOME_TOKEN": "ghp_abcdef1234567890"}, + } + } + ) + problems = c.validate_catalog(data) + assert any("real secret value" in p for p in problems) + + +def test_validate_catalog_rejects_node_options_env_walking_past_allowlist(): + # The exact reproduction from issue #68 finding 2: an allowlisted + # `npx` command carrying NODE_OPTIONS in env, which previously passed + # validation and would have flowed straight into the executed + # subprocess via catalog_entry_to_paste_json(). + data = _catalog_with( + { + "config": { + "command": "npx", + "args": ["-y", "widget-mcp@1.0.0"], + "env": {"NODE_OPTIONS": "--require /tmp/payload.js"}, + } + } + ) + problems = c.validate_catalog(data) + assert problems != [] + + +# --- validate_catalog: version pinning (issue #68 finding 3) --------------- # +def test_validate_catalog_rejects_unpinned_npx_package(): + data = _catalog_with({"config": {"command": "npx", "args": ["-y", "widget-mcp"]}}) + problems = c.validate_catalog(data) + assert any("not version-pinned" in p for p in problems) + + +def test_validate_catalog_rejects_unpinned_scoped_npx_package(): + data = _catalog_with({"config": {"command": "npx", "args": ["-y", "@scope/pkg"]}}) + problems = c.validate_catalog(data) + assert any("not version-pinned" in p for p in problems) + + +def test_validate_catalog_accepts_pinned_scoped_npx_package(): + data = _catalog_with({"config": {"command": "npx", "args": ["-y", "@scope/pkg@1.2.3"]}}) + assert c.validate_catalog(data) == [] + + +def test_validate_catalog_rejects_unpinned_uvx_package(): + data = _catalog_with({"config": {"command": "uvx", "args": ["some-tool"]}}) + problems = c.validate_catalog(data) + assert any("not version-pinned" in p for p in problems) + + +def test_validate_catalog_accepts_uvx_at_version_pin(): + data = _catalog_with({"config": {"command": "uvx", "args": ["some-tool@1.0.0"]}}) + assert c.validate_catalog(data) == [] + + +def test_validate_catalog_accepts_uvx_double_equals_pin(): + data = _catalog_with({"config": {"command": "uvx", "args": ["some-tool==1.0.0"]}}) + assert c.validate_catalog(data) == [] + + +def test_validate_catalog_rejects_docker_latest_tag(): + data = _catalog_with({"config": {"command": "docker", "args": ["run", "some/image:latest"]}}) + problems = c.validate_catalog(data) + assert any("'latest'" in p for p in problems) + + +def test_validate_catalog_rejects_docker_untagged_image(): + data = _catalog_with({"config": {"command": "docker", "args": ["run", "some/image"]}}) + problems = c.validate_catalog(data) + assert any("no explicit tag" in p for p in problems) + + +def test_validate_catalog_accepts_pinned_docker_image_with_flags(): + data = _catalog_with( + { + "config": { + "command": "docker", + "args": ["run", "-i", "--rm", "-e", "SOME_TOKEN", "some/image:1.2.3"], + } + } + ) + assert c.validate_catalog(data) == [] + + +def test_validate_catalog_does_not_pin_check_placeholders_flags_or_subcommand(): + # A pinned uvx spec followed by flags and a positional + # must not itself get mistaken for an unpinned package. + data = _catalog_with( + { + "config": { + "command": "uvx", + "args": ["mcp-server-git@2026.7.10", "--repository", ""], + } + } + ) + assert c.validate_catalog(data) == [] + + +# --- validate_catalog: id constraint (issue #68 finding 7) ----------------- # +def test_validate_catalog_rejects_id_with_markup(): + data = _catalog_with({"id": "Verified"}) + problems = c.validate_catalog(data) + assert any("must match" in p for p in problems) + + +def test_validate_catalog_rejects_id_with_uppercase(): + data = _catalog_with({"id": "Widget"}) + problems = c.validate_catalog(data) + assert any("must match" in p for p in problems) + + +def test_validate_catalog_rejects_id_starting_with_dash(): + data = _catalog_with({"id": "-widget"}) + problems = c.validate_catalog(data) + assert any("must match" in p for p in problems) + + +def test_validate_catalog_accepts_valid_slug_id(): + data = _catalog_with({"id": "widget-2.thing-ok"}) + assert c.validate_catalog(data) == [] + + # --- resolve_catalog -------------------------------------------------------- # def test_resolve_catalog_nothing_available_returns_empty_dict(): assert c.resolve_catalog(None, None, None) == {} @@ -2003,13 +2214,93 @@ def test_resolve_catalog_rejects_absurd_version_jump(monkeypatch): pub = priv.public_key().public_bytes_raw() monkeypatch.setattr(c, "CATALOG_PUBKEYS", [pub]) - cached = _signed(_minimal_catalog(version=5), priv) + # The freeze attempt goes FIRST (as `cached`), the legitimate catalog + # SECOND (as `remote`) -- on purpose. Putting the good catalog first + # (as an earlier version of this test did) never exercises the + # vulnerable path: the old implementation only guarded a candidate + # against "the best accepted so far," so whichever candidate was + # evaluated FIRST got in unconditionally, uncapped. Ordering the freeze + # attempt first is what actually proves the cap holds regardless of + # evaluation order. freeze_attempt = _signed(_minimal_catalog(version=999999), priv) + good = _signed(_minimal_catalog(version=5), priv) - result = c.resolve_catalog(None, cached, freeze_attempt) + result = c.resolve_catalog(None, freeze_attempt, good) assert c.catalog_version(result) == 5 +def test_resolve_catalog_caps_first_and_only_candidate(monkeypatch): + # issue #68 finding 6: with no bundled catalog to anchor against, a + # signed catalog claiming an absurd version must still be capped even + # when it is the ONLY candidate resolve_catalog() ever sees -- there is + # no "best so far" for it to be compared against, so the cap has to + # apply unconditionally, not "once something else has already landed." + priv = Ed25519PrivateKey.generate() + pub = priv.public_key().public_bytes_raw() + monkeypatch.setattr(c, "CATALOG_PUBKEYS", [pub]) + + freeze_attempt = _signed(_minimal_catalog(version=999999999), priv) + + result = c.resolve_catalog(None, None, freeze_attempt) + assert result == {} + + +def test_resolve_catalog_anchors_cap_to_bundled_not_a_chained_best(monkeypatch): + # Anti-freeze must be measured against the BUNDLED version specifically, + # not against "whatever the best-so-far happens to be after each + # candidate is accepted" -- a chained anchor lets each accepted + # candidate ratchet the allowed ceiling upward, so a legitimate + # moderate bump (cached) plus a second, much larger jump (remote) can + # each individually look "within _CATALOG_MAX_VERSION_JUMP of the + # previous one" while remote is nowhere near bundled's version. + priv = Ed25519PrivateKey.generate() + pub = priv.public_key().public_bytes_raw() + monkeypatch.setattr(c, "CATALOG_PUBKEYS", [pub]) + + bundled = _signed(_minimal_catalog(version=2), priv) + cached = _signed(_minimal_catalog(version=1000), priv) # within 1000 of bundled + remote = _signed(_minimal_catalog(version=1900), priv) # within 1000 of cached, + # NOT of bundled + + result = c.resolve_catalog(bundled, cached, remote) + assert c.catalog_version(result) == 1000 + + +def test_resolve_catalog_prefers_bundled_on_version_tie(monkeypatch): + # issue #68 finding 6: on a tie the LAST candidate evaluated used to + # win, so remote silently beat bundled at equal version. Bundled -- + # the copy frozen into the binary -- must win ties. + priv = Ed25519PrivateKey.generate() + pub = priv.public_key().public_bytes_raw() + monkeypatch.setattr(c, "CATALOG_PUBKEYS", [pub]) + + bundled_data = _minimal_catalog(version=5) + bundled = _signed(bundled_data, priv) + + remote_data = _minimal_catalog(version=5) + remote_data["servers"][0]["display"] = "Remote Impostor" + remote = _signed(remote_data, priv) + + result = c.resolve_catalog(bundled, None, remote) + assert c.catalog_version(result) == 5 + assert result["servers"][0]["display"] == "Widget" + + +def test_resolve_catalog_floor_rejects_below_persisted_version(monkeypatch): + # `floor` is a pure parameter: the caller (eventually the GUI, from + # persisted storage) can pass a previously-accepted version, and + # nothing below it may be accepted even with no bundled catalog to + # anchor against. + priv = Ed25519PrivateKey.generate() + pub = priv.public_key().public_bytes_raw() + monkeypatch.setattr(c, "CATALOG_PUBKEYS", [pub]) + + stale = _signed(_minimal_catalog(version=3), priv) + + result = c.resolve_catalog(None, None, stale, floor=10) + assert result == {} + + def test_resolve_catalog_malformed_candidate_does_not_raise(monkeypatch): priv = Ed25519PrivateKey.generate() pub = priv.public_key().public_bytes_raw() @@ -2039,15 +2330,32 @@ def test_resolve_catalog_invalid_but_signed_candidate_is_skipped(monkeypatch): def test_catalog_entry_to_paste_json_basic_shape(): entry = _minimal_catalog()["servers"][0] result = c.catalog_entry_to_paste_json(entry) - assert result == {"widget": {"command": "npx", "args": ["-y", "widget-mcp"]}} + assert result == {"widget": {"command": "npx", "args": ["-y", "widget-mcp@1.0.0"]}} def test_catalog_entry_to_paste_json_includes_env_when_present(): + # NOTE: catalog_entry_to_paste_json() is a pure shape-converter for an + # entry that has ALREADY passed validate_catalog() -- it is correct for + # it to carry env through verbatim. The bug (issue #68 finding 2) was + # never in this function; it was that validate_catalog() let entries + # with dangerous/non-placeholder env values reach this function in the + # first place. This test now proves that boundary explicitly: a + # validation-legal env value (a token) survives the + # conversion, and a value validate_catalog() would have rejected is + # confirmed rejected before it ever gets here. entry = _minimal_catalog()["servers"][0] entry["config"]["env"] = {"GRAFANA_URL": ""} result = c.catalog_entry_to_paste_json(entry) assert result["widget"]["env"] == {"GRAFANA_URL": ""} + catalog = _minimal_catalog() + catalog["servers"][0]["config"]["env"] = {"GRAFANA_URL": ""} + assert c.validate_catalog(catalog) == [] + + malicious = _minimal_catalog() + malicious["servers"][0]["config"]["env"] = {"NODE_OPTIONS": "--require /tmp/payload.js"} + assert c.validate_catalog(malicious) != [] + def test_config_has_unfilled_placeholders_true_for_token(): cfg = {"command": "npx", "args": ["-y", "server", ""]}