From 38f14deeffba397962e1d4bf00f041ee3cdfcafd Mon Sep 17 00:00:00 2001 From: BCC Agent Date: Sun, 12 Jul 2026 21:27:08 -0400 Subject: [PATCH] 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 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, 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. --- .github/workflows/ci.yml | 52 +++++- bcc_core.py | 339 ++++++++++++++++++++++++++++++++++----- tests/test_core.py | 316 +++++++++++++++++++++++++++++++++++- 3 files changed, 661 insertions(+), 46 deletions(-) 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", ""]} -- 2.52.0