From 8c4985601cb4132e6d1f3cf17510890ef240d33e Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Fri, 17 Jul 2026 02:38:26 -0400 Subject: [PATCH] feat(mcp): add registry posture scanning Signed-off-by: Rod Boev --- src/skillspector/cli.py | 35 ++ src/skillspector/mcp_registry.py | 429 ++++++++++++++++++ tests/fixtures/mcp_registry/malformed.json | 1 + tests/fixtures/mcp_registry/mcp_registry.json | 52 +++ tests/unit/test_cli.py | 66 +++ tests/unit/test_mcp_registry.py | 421 +++++++++++++++++ 6 files changed, 1004 insertions(+) create mode 100644 src/skillspector/mcp_registry.py create mode 100644 tests/fixtures/mcp_registry/malformed.json create mode 100644 tests/fixtures/mcp_registry/mcp_registry.json create mode 100644 tests/unit/test_mcp_registry.py diff --git a/src/skillspector/cli.py b/src/skillspector/cli.py index ec52cc76..daf0a3b9 100644 --- a/src/skillspector/cli.py +++ b/src/skillspector/cli.py @@ -37,6 +37,7 @@ from skillspector.constants import RISK_THRESHOLD from skillspector.graph import graph from skillspector.logging_config import get_logger, set_level +from skillspector.mcp_registry import scan_registry from skillspector.multi_skill import MultiSkillDetectionResult, detect_skills from skillspector.suppression import build_baseline_dict, dump_baseline, load_baseline @@ -253,6 +254,13 @@ def scan( help="Show detailed progress.", ), ] = False, + mcp_registry: Annotated[ + bool, + typer.Option( + "--mcp-registry", + help="Scan an MCP Registry payload or URL instead of a skill.", + ), + ] = False, ) -> None: """ Scan a skill for security vulnerabilities. @@ -284,6 +292,33 @@ def scan( chain when unset; AWS_REGION default: us-west-2) NVIDIA_INFERENCE_KEY for the NVIDIA providers """ + if mcp_registry: + if recursive or baseline is not None or show_suppressed or yara_rules_dir is not None: + console.print( + "[red]Error:[/red] --mcp-registry cannot be combined with " + "--recursive, --baseline, --show-suppressed, or --yara-rules-dir" + ) + raise typer.Exit(code=2) + if format != FormatChoice.json: + console.print("[red]Error:[/red] --mcp-registry currently supports only --format json") + raise typer.Exit(code=2) + try: + result = scan_registry(input_path) + report = json.dumps(result, indent=2) + if output: + output.write_text(report, encoding="utf-8") + console.print(f"Report saved to: {output}") + else: + print(report) + if result["risk_score"] > RISK_THRESHOLD: + raise typer.Exit(code=1) + except typer.Exit: + raise + except Exception as e: + console.print(f"[red]Error:[/red] {e}") + raise typer.Exit(code=2) from e + return + if verbose: set_level("DEBUG") diff --git a/src/skillspector/mcp_registry.py b/src/skillspector/mcp_registry.py new file mode 100644 index 00000000..5678cda5 --- /dev/null +++ b/src/skillspector/mcp_registry.py @@ -0,0 +1,429 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""MCP Registry acquisition, normalized snapshots, and posture checks.""" + +from __future__ import annotations + +import hashlib +import json +import re +from dataclasses import asdict, dataclass +from datetime import UTC, datetime +from itertools import chain +from pathlib import Path +from typing import Any, TypedDict + +import httpx + +REGISTRY_URL = "https://registry.modelcontextprotocol.io/v0/servers" +OFFICIAL_META_KEY = "io.modelcontextprotocol.registry/official" +FILE_SHA256_RE = re.compile(r"^[a-f0-9]{64}$") +MUTABLE_VERSION_TAGS = frozenset( + { + "latest", + "next", + "beta", + "alpha", + "stable", + "canary", + "edge", + "main", + "master", + "dev", + "nightly", + "preview", + } +) +RANGE_SYNTAX_RE = re.compile(r"[\^~*><=|]|\s") +WILDCARD_SEGMENT_RE = re.compile(r"(?:^|\.)[xX*](?:\.|$)") +NPM_EXACT_VERSION_RE = re.compile(r"^v?\d+\.\d+\.\d+(?:-[0-9A-Za-z.-]+)?(?:\+[0-9A-Za-z.-]+)?$") + + +class RegistryFinding(TypedDict): + id: str + target: str + message: str + severity: str + evidence: str + risk_score: int + + +class RegistryServerReport(TypedDict): + snapshot: dict[str, Any] + findings: list[RegistryFinding] + + +@dataclass(frozen=True) +class RepositoryReference: + url: str | None = None + source: str | None = None + id: str | None = None + subfolder: str | None = None + + +@dataclass(frozen=True) +class PackageReference: + registry_type: str | None = None + identifier: str | None = None + version: str | None = None + file_sha256: str | None = None + transport_type: str | None = None + transport_url: str | None = None + + +@dataclass(frozen=True) +class RemoteReference: + type: str | None = None + url: str | None = None + + +@dataclass(frozen=True) +class RegistryServerSnapshot: + source: str + name: str + title: str | None + description: str | None + version: str | None + website_url: str | None + repository: RepositoryReference | None + packages: tuple[PackageReference, ...] + remotes: tuple[RemoteReference, ...] + status: str | None + published_at: str | None + updated_at: str | None + is_latest: bool | None + record_hash: str + scanned_at: str + + def to_dict(self) -> dict[str, Any]: + data = asdict(self) + data["packages"] = [asdict(package) for package in self.packages] + data["remotes"] = [asdict(remote) for remote in self.remotes] + return data + + +def _canonical_json(value: Any) -> str: + return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + + +def record_hash(record: dict[str, Any]) -> str: + """Hash a normalized owner record independently of JSON object key order.""" + return hashlib.sha256(_canonical_json(record).encode("utf-8")).hexdigest() + + +def _optional_string(value: Any) -> str | None: + # The registry owns field semantics; non-string values are recorded as + # absent so checks report unavailable evidence instead of failing the scan. + return value if isinstance(value, str) else None + + +def _official_meta(record: dict[str, Any]) -> dict[str, Any]: + meta = record.get("_meta", {}) + official = meta.get(OFFICIAL_META_KEY, {}) if isinstance(meta, dict) else {} + return official if isinstance(official, dict) else {} + + +def _is_specific_package_version(registry_type: str | None, version: str | None) -> bool: + if version is None: + return False + if version.casefold() in MUTABLE_VERSION_TAGS: + return False + if not any(char.isdigit() for char in version): + return False + if registry_type == "npm": + return NPM_EXACT_VERSION_RE.fullmatch(version) is not None + # Prerelease/build suffixes like 1.0.0-linux-x64 are exact versions; only + # range operators and whole x/* segments (1.x, 1.*) mark a mutable range. + return not (RANGE_SYNTAX_RE.search(version) or WILDCARD_SEGMENT_RE.search(version)) + + +def _is_valid_file_sha256(file_sha256: str | None) -> bool: + return file_sha256 is not None and FILE_SHA256_RE.fullmatch(file_sha256) is not None + + +def _record_dict_list( + record: dict[str, Any], field_name: str, *, source: str, server_name: str +) -> list[dict[str, Any]]: + if field_name not in record: + return [] + value = record[field_name] + if not isinstance(value, list) or any(not isinstance(item, dict) for item in value): + raise ValueError( + f"MCP Registry payload has an invalid {field_name} collection for {server_name} from {source}" + ) + return value + + +def _normalize_package_reference(package: dict[str, Any]) -> PackageReference: + transport = package.get("transport") + transport = transport if isinstance(transport, dict) else {} + return PackageReference( + registry_type=_optional_string(package.get("registryType")), + identifier=_optional_string(package.get("identifier")), + version=_optional_string(package.get("version")), + file_sha256=_optional_string(package.get("fileSha256")), + transport_type=_optional_string(transport.get("type")), + transport_url=_optional_string(transport.get("url")), + ) + + +def normalize_server( + entry: dict[str, Any], *, source: str, scanned_at: str | None = None +) -> RegistryServerSnapshot: + if not isinstance(entry, dict) or not isinstance(entry.get("server"), dict): + raise ValueError(f"MCP Registry payload has an invalid server record from {source}") + record = entry["server"] + name = _optional_string(record.get("name")) + if not name: + raise ValueError(f"MCP Registry payload has a server without a name from {source}") + repository_data = record.get("repository") + repository = None + if repository_data is not None and not isinstance(repository_data, dict): + raise ValueError( + f"MCP Registry payload has an invalid repository object for {name} from {source}" + ) + if isinstance(repository_data, dict): + repository = RepositoryReference( + url=_optional_string(repository_data.get("url")), + source=_optional_string(repository_data.get("source")), + id=_optional_string(repository_data.get("id")), + subfolder=_optional_string(repository_data.get("subfolder")), + ) + packages = tuple( + _normalize_package_reference(package) + for package in _record_dict_list(record, "packages", source=source, server_name=name) + ) + remotes = tuple( + RemoteReference( + type=_optional_string(remote.get("type")), + url=_optional_string(remote.get("url")), + ) + for remote in _record_dict_list(record, "remotes", source=source, server_name=name) + ) + official = _official_meta(entry) + return RegistryServerSnapshot( + source=source, + name=name, + title=_optional_string(record.get("title")), + description=_optional_string(record.get("description")), + version=_optional_string(record.get("version")), + website_url=_optional_string(record.get("websiteUrl")), + repository=repository, + packages=packages, + remotes=remotes, + status=_optional_string(official.get("status")), + published_at=_optional_string(official.get("publishedAt")), + updated_at=_optional_string(official.get("updatedAt")), + is_latest=official.get("isLatest") if isinstance(official.get("isLatest"), bool) else None, + record_hash=record_hash({"server": record, OFFICIAL_META_KEY: official}), + scanned_at=scanned_at or datetime.now(UTC).isoformat(), + ) + + +def normalize_payload(payload: dict[str, Any], *, source: str) -> list[RegistryServerSnapshot]: + if not isinstance(payload, dict) or not isinstance(payload.get("servers"), list): + raise ValueError(f"MCP Registry payload from {source} must contain a servers list") + scanned_at = datetime.now(UTC).isoformat() + return [ + normalize_server(entry, source=source, scanned_at=scanned_at) + for entry in payload["servers"] + ] + + +def _finding( + rule: str, + message: str, + target: str, + *, + severity: str, + evidence: str, + risk_score: int, +) -> RegistryFinding: + return { + "id": rule, + "target": target, + "message": message, + "severity": severity, + "evidence": evidence, + "risk_score": risk_score, + } + + +def _unavailable(rule: str, message: str, target: str) -> RegistryFinding: + return _finding( + rule, + message, + target, + severity="info", + evidence="unavailable", + risk_score=0, + ) + + +def _registry_assertion( + rule: str, message: str, target: str, *, severity: str, risk_score: int +) -> RegistryFinding: + return _finding( + rule, + message, + target, + severity=severity, + evidence="registry_assertion", + risk_score=risk_score, + ) + + +def posture_findings(snapshot: RegistryServerSnapshot) -> list[RegistryFinding]: + findings: list[RegistryFinding] = [] + for index, package in enumerate(snapshot.packages): + target = package.identifier or f"package[{index}]" + if package.version is None: + findings.append( + _unavailable("MCP-PACKAGE-VERSION", "Package version is unavailable", target) + ) + elif not _is_specific_package_version(package.registry_type, package.version): + findings.append( + _registry_assertion( + "MCP-PACKAGE-VERSION", + "Package version is not pinned", + target, + severity="high", + risk_score=30, + ) + ) + if package.file_sha256 is None: + findings.append( + _unavailable("MCP-PACKAGE-SHA256", "Package fileSha256 is unavailable", target) + ) + elif not _is_valid_file_sha256(package.file_sha256): + findings.append( + _registry_assertion( + "MCP-PACKAGE-SHA256", + "Package fileSha256 is invalid", + target, + severity="high", + risk_score=25, + ) + ) + if snapshot.repository is None or not snapshot.repository.url: + findings.append( + _unavailable("MCP-REPOSITORY", "Repository reference is unavailable", snapshot.name) + ) + if snapshot.status is None: + findings.append( + _unavailable("MCP-OFFICIAL-STATUS", "Official status is unavailable", snapshot.name) + ) + elif snapshot.status != "active": + findings.append( + _registry_assertion( + "MCP-OFFICIAL-STATUS", + f"Official status is {snapshot.status}", + snapshot.name, + severity="medium", + risk_score=20, + ) + ) + for remote in snapshot.remotes: + if remote.url and remote.url.lower().startswith("http://"): + findings.append( + _registry_assertion( + "MCP-PLAIN-HTTP", + "Remote endpoint uses plain HTTP", + remote.url, + severity="high", + risk_score=25, + ) + ) + return findings + + +def _dict_payload(payload: object, *, source: str) -> dict[str, Any]: + if not isinstance(payload, dict): + raise ValueError(f"MCP Registry source failed: {source}: payload must be a JSON object") + return payload + + +def _load_payload(input_path: str) -> dict[str, Any]: + source = input_path + try: + if Path(input_path).is_file(): + return _dict_payload( + json.loads(Path(input_path).read_text(encoding="utf-8")), + source=source, + ) + if input_path.startswith(("http://", "https://")): + if input_path != REGISTRY_URL: + raise ValueError( + f"MCP Registry source failed: {input_path}: only the official registry URL is supported" + ) + return _load_paginated_registry(input_path) + payload = _load_paginated_registry(REGISTRY_URL) + matches = [ + entry + for entry in payload.get("servers", []) + if isinstance(entry, dict) + and isinstance(entry.get("server"), dict) + and entry["server"].get("name") == input_path + ] + if not matches: + raise ValueError(f"MCP Registry server identifier was not found: {source}") + # The registry lists every published version of a server; a name scan + # assesses the owner's latest record, not the historical tail. + latest = [entry for entry in matches if _official_meta(entry).get("isLatest") is True] + return {"servers": latest or matches} + except (OSError, json.JSONDecodeError, httpx.HTTPError, ValueError) as exc: + if isinstance(exc, ValueError) and str(exc).startswith("MCP Registry source"): + raise + raise ValueError(f"MCP Registry source failed: {source}: {exc}") from exc + + +def _load_paginated_registry(url: str) -> dict[str, Any]: + pages: list[dict[str, Any]] = [] + seen_cursors: set[str] = set() + cursor: str | None = None + + while True: + params = {"cursor": cursor} if cursor is not None else None + response = httpx.get(url, params=params, timeout=30) + response.raise_for_status() + payload = _dict_payload(response.json(), source=url) + if not isinstance(payload.get("servers"), list): + raise ValueError(f"MCP Registry payload from {url} must contain a servers list") + pages.append(payload) + + metadata = payload.get("metadata") + next_cursor = metadata.get("nextCursor") if isinstance(metadata, dict) else None + if not isinstance(next_cursor, str) or not next_cursor: + break + if next_cursor in seen_cursors: + raise ValueError(f"MCP Registry source failed: {url}: repeated pagination cursor") + seen_cursors.add(next_cursor) + cursor = next_cursor + + return { + "servers": list(chain.from_iterable(page["servers"] for page in pages)), + "metadata": pages[-1].get("metadata", {}), + } + + +def scan_registry(input_path: str = REGISTRY_URL) -> dict[str, Any]: + """Acquire, normalize, and assess one MCP Registry payload.""" + snapshots = normalize_payload(_load_payload(input_path), source=input_path) + per_server: list[RegistryServerReport] = [ + {"snapshot": snapshot.to_dict(), "findings": posture_findings(snapshot)} + for snapshot in snapshots + ] + findings = [finding for server in per_server for finding in server["findings"]] + risk_score = min(sum(finding["risk_score"] for finding in findings), 100) + max_risk_score = max((finding["risk_score"] for finding in findings), default=0) + return { + "mcp_registry": True, + "source": input_path, + "server_count": len(snapshots), + "risk_score": risk_score, + "max_risk_score": max_risk_score, + "findings": findings, + "snapshots": [snapshot.to_dict() for snapshot in snapshots], + "servers": per_server, + } diff --git a/tests/fixtures/mcp_registry/malformed.json b/tests/fixtures/mcp_registry/malformed.json new file mode 100644 index 00000000..d443d55e --- /dev/null +++ b/tests/fixtures/mcp_registry/malformed.json @@ -0,0 +1 @@ +{"servers": [{"not_server": {}}]} diff --git a/tests/fixtures/mcp_registry/mcp_registry.json b/tests/fixtures/mcp_registry/mcp_registry.json new file mode 100644 index 00000000..3d6097c6 --- /dev/null +++ b/tests/fixtures/mcp_registry/mcp_registry.json @@ -0,0 +1,52 @@ +{ + "servers": [ + { + "server": { + "$schema": "https://static.modelcontextprotocol.io/schemas/2025-12-11/server.schema.json", + "name": "ac.tandem/docs-mcp", + "title": "Tandem Docs", + "description": "Remote MCP server for Tandem docs.", + "version": "0.3.2", + "websiteUrl": "https://tandem.ac/docs-mcp", + "repository": {"url": "https://github.com/frumu-ai/tandem", "source": "github"}, + "remotes": [{"type": "streamable-http", "url": "https://tandem.ac/mcp"}] + }, + "_meta": { + "io.modelcontextprotocol.registry/official": { + "status": "active", + "publishedAt": "2026-04-22T21:06:34.500049Z", + "updatedAt": "2026-04-22T21:06:34.500049Z", + "isLatest": true + } + } + }, + { + "server": { + "$schema": "https://static.modelcontextprotocol.io/schemas/2025-12-11/server.schema.json", + "name": "ai.adeu/adeu", + "description": "Automated DOCX Redlining Engine", + "repository": {"url": "https://github.com/dealfluence/adeu", "source": "github"}, + "version": "1.7.1", + "packages": [{ + "registryType": "npm", + "identifier": "@adeu/mcp-server", + "version": "1.7.1", + "fileSha256": "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef", + "transport": {"type": "stdio"} + }] + }, + "_meta": {"io.modelcontextprotocol.registry/official": {"status": "active", "isLatest": true}} + }, + { + "server": { + "$schema": "https://static.modelcontextprotocol.io/schemas/2025-12-11/server.schema.json", + "name": "ai.agenticshelf/graffeo", + "title": "Graffeo Coffee Roasting", + "version": "1.0.0", + "remotes": [{"type": "streamable-http", "url": "http://example.invalid/mcp"}] + }, + "_meta": {"io.modelcontextprotocol.registry/official": {"status": "deprecated", "isLatest": false}} + } + ], + "metadata": {"count": 3} +} diff --git a/tests/unit/test_cli.py b/tests/unit/test_cli.py index e340ccd8..92143893 100644 --- a/tests/unit/test_cli.py +++ b/tests/unit/test_cli.py @@ -75,6 +75,72 @@ def test_cli_scan_nonexistent_exits_2() -> None: assert "Error" in result.output or "error" in result.output.lower() +def test_cli_mcp_registry_routes_and_writes_json(tmp_path: Path) -> None: + payload = tmp_path / "registry.json" + payload.write_text('{"servers": []}', encoding="utf-8") + output = tmp_path / "registry-report.json" + result = runner.invoke( + app, + ["scan", str(payload), "--mcp-registry", "--format", "json", "--output", str(output)], + ) + assert result.exit_code == 0 + assert json.loads(output.read_text(encoding="utf-8"))["mcp_registry"] is True + + +def test_cli_mcp_registry_exits_1_when_aggregate_risk_crosses_threshold(tmp_path: Path) -> None: + payload = tmp_path / "registry.json" + payload.write_text( + json.dumps( + { + "servers": [ + { + "server": { + "name": "risky/example", + "remotes": [ + {"type": "streamable-http", "url": "http://one.invalid/mcp"}, + {"type": "streamable-http", "url": "http://two.invalid/mcp"}, + {"type": "streamable-http", "url": "http://three.invalid/mcp"}, + ], + }, + "_meta": { + "io.modelcontextprotocol.registry/official": {"status": "deprecated"} + }, + } + ] + } + ), + encoding="utf-8", + ) + result = runner.invoke(app, ["scan", str(payload), "--mcp-registry", "--format", "json"]) + assert result.exit_code == 1 + assert json.loads(result.output)["risk_score"] == 95 + + +@pytest.mark.parametrize( + "args", [[], ["--format", "terminal"], ["--format", "markdown"], ["--format", "sarif"]] +) +def test_cli_mcp_registry_rejects_non_json_formats(tmp_path: Path, args: list[str]) -> None: + payload = tmp_path / "registry.json" + payload.write_text('{"servers": []}', encoding="utf-8") + result = runner.invoke(app, ["scan", str(payload), "--mcp-registry", *args]) + assert result.exit_code == 2 + assert "supports only --format json" in result.output + + +@pytest.mark.parametrize( + "flag", ["--recursive", "--baseline", "--show-suppressed", "--yara-rules-dir"] +) +def test_cli_mcp_registry_rejects_skill_only_flags(tmp_path: Path, flag: str) -> None: + payload = tmp_path / "registry.json" + payload.write_text('{"servers": []}', encoding="utf-8") + args = ["scan", str(payload), "--mcp-registry", flag] + if flag in {"--baseline", "--yara-rules-dir"}: + args.append(str(tmp_path / "value")) + result = runner.invoke(app, args) + assert result.exit_code == 2 + assert "cannot be combined" in result.output + + def test_cli_scan_missing_baseline_exits_2(tmp_path: Path) -> None: """scan with a --baseline pointing at a missing file exits with code 2.""" (tmp_path / "SKILL.md").write_text("# Hi", encoding="utf-8") diff --git a/tests/unit/test_mcp_registry.py b/tests/unit/test_mcp_registry.py new file mode 100644 index 00000000..8dac6aa1 --- /dev/null +++ b/tests/unit/test_mcp_registry.py @@ -0,0 +1,421 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Tests for the MCP Registry owner and posture checks.""" + +import json +from pathlib import Path +from typing import Any + +import httpx +import pytest + +from skillspector.mcp_registry import ( + OFFICIAL_META_KEY, + REGISTRY_URL, + normalize_payload, + posture_findings, + record_hash, + scan_registry, +) + +FIXTURES = Path(__file__).parents[1] / "fixtures" / "mcp_registry" + + +def payload() -> dict: + return json.loads((FIXTURES / "mcp_registry.json").read_text(encoding="utf-8")) + + +def one_server(server: dict[str, Any], official: dict[str, Any] | None = None) -> dict[str, Any]: + entry: dict[str, Any] = {"server": server} + if official is not None: + entry["_meta"] = {OFFICIAL_META_KEY: official} + return {"servers": [entry]} + + +def pinned_package(**overrides: Any) -> dict[str, Any]: + package: dict[str, Any] = { + "registryType": "npm", + "identifier": "example", + "version": "1.0.0", + "fileSha256": "a" * 64, + "transport": {"type": "stdio"}, + } + package.update(overrides) + return package + + +def pinned_server(**overrides: Any) -> dict[str, Any]: + server: dict[str, Any] = { + "name": "safe/example", + "repository": {"url": "https://github.com/example/project", "source": "github"}, + "packages": [pinned_package()], + "remotes": [{"type": "streamable-http", "url": "https://example.invalid/mcp"}], + } + server.update(overrides) + return server + + +class _Response: + def __init__(self, payload: dict[str, Any]) -> None: + self._payload = payload + + def raise_for_status(self) -> None: + return None + + def json(self) -> dict[str, Any]: + return self._payload + + +def test_snapshot_normalizes_owner_fields_and_serializes() -> None: + first = normalize_payload(payload(), source="fixture")[0] + data = first.to_dict() + assert data.pop("record_hash") == first.record_hash + assert data.pop("scanned_at") == first.scanned_at + assert data == { + "source": "fixture", + "name": "ac.tandem/docs-mcp", + "title": "Tandem Docs", + "description": "Remote MCP server for Tandem docs.", + "version": "0.3.2", + "website_url": "https://tandem.ac/docs-mcp", + "repository": { + "url": "https://github.com/frumu-ai/tandem", + "source": "github", + "id": None, + "subfolder": None, + }, + "packages": [], + "remotes": [{"type": "streamable-http", "url": "https://tandem.ac/mcp"}], + "status": "active", + "published_at": "2026-04-22T21:06:34.500049Z", + "updated_at": "2026-04-22T21:06:34.500049Z", + "is_latest": True, + } + + +def test_snapshot_preserves_package_transport() -> None: + server = pinned_server( + packages=[ + pinned_package( + transport={"type": "streamable-http", "url": "https://example.invalid/mcp"} + ) + ] + ) + package = normalize_payload(one_server(server), source="fixture")[0].packages[0] + assert package.transport_type == "streamable-http" + assert package.transport_url == "https://example.invalid/mcp" + + +def test_snapshot_preserves_template_transport_url() -> None: + server = pinned_server( + packages=[pinned_package(transport={"type": "streamable-http", "url": "{baseUrl}/mcp"})] + ) + snapshot = normalize_payload( + one_server(server, official={"status": "active"}), source="fixture" + )[0] + assert snapshot.packages[0].transport_url == "{baseUrl}/mcp" + assert posture_findings(snapshot) == [] + + +def test_snapshot_treats_wrong_typed_optional_fields_as_absent() -> None: + server = pinned_server(packages=[pinned_package(version=7, fileSha256=7)]) + snapshot = normalize_payload(one_server(server, official={"status": 0}), source="fixture")[0] + package = snapshot.packages[0] + assert package.version is None + assert package.file_sha256 is None + assert snapshot.status is None + assert all(finding["evidence"] == "unavailable" for finding in posture_findings(snapshot)) + + +def test_contract_isolation_uses_normalized_snapshots() -> None: + report = scan_registry(str(FIXTURES / "mcp_registry.json")) + assert report["mcp_registry"] is True + assert report["snapshots"][0]["repository"]["url"].startswith("https://") + assert all("server" not in server["snapshot"] for server in report["servers"]) + + +def test_registry_url_scan_follows_next_cursor(monkeypatch: pytest.MonkeyPatch) -> None: + page_one = { + "servers": [{"server": {"name": "page/one"}}], + "metadata": {"nextCursor": "cursor-2"}, + } + page_two = {"servers": [{"server": {"name": "page/two"}}], "metadata": {}} + calls: list[tuple[str, dict[str, str] | None]] = [] + + def fake_get(url: str, *, params: dict[str, str] | None = None, timeout: int) -> _Response: + calls.append((url, params)) + return _Response(page_one if params is None else page_two) + + monkeypatch.setattr("skillspector.mcp_registry.httpx.get", fake_get) + + report = scan_registry(REGISTRY_URL) + + assert report["server_count"] == 2 + assert [server["snapshot"]["name"] for server in report["servers"]] == ["page/one", "page/two"] + assert calls == [(REGISTRY_URL, None), (REGISTRY_URL, {"cursor": "cursor-2"})] + + +def test_untrusted_registry_url_is_rejected() -> None: + with pytest.raises(ValueError, match="only the official registry URL is supported"): + scan_registry("https://untrusted.invalid/servers") + + +def test_server_identifier_scan_follows_next_cursor(monkeypatch: pytest.MonkeyPatch) -> None: + page_one = { + "servers": [{"server": {"name": "page/one"}}], + "metadata": {"nextCursor": "cursor-2"}, + } + page_two = {"servers": [{"server": {"name": "page/two"}}], "metadata": {}} + + def fake_get(url: str, *, params: dict[str, str] | None = None, timeout: int) -> _Response: + return _Response(page_one if params is None else page_two) + + monkeypatch.setattr("skillspector.mcp_registry.httpx.get", fake_get) + + report = scan_registry("page/two") + + assert report["server_count"] == 1 + assert report["servers"][0]["snapshot"]["name"] == "page/two" + + +def test_server_identifier_scan_selects_latest_version(monkeypatch: pytest.MonkeyPatch) -> None: + page = { + "servers": [ + { + "server": {"name": "dup/example", "version": "1.0.0"}, + "_meta": {OFFICIAL_META_KEY: {"status": "deprecated", "isLatest": False}}, + }, + { + "server": {"name": "dup/example", "version": "1.1.0"}, + "_meta": {OFFICIAL_META_KEY: {"status": "active", "isLatest": True}}, + }, + ], + "metadata": {}, + } + + def fake_get(url: str, *, params: dict[str, str] | None = None, timeout: int) -> _Response: + return _Response(page) + + monkeypatch.setattr("skillspector.mcp_registry.httpx.get", fake_get) + + report = scan_registry("dup/example") + + assert report["server_count"] == 1 + assert report["servers"][0]["snapshot"]["version"] == "1.1.0" + assert all(finding["id"] != "MCP-OFFICIAL-STATUS" for finding in report["findings"]) + + +def test_record_hash_is_stable_when_record_keys_are_reordered() -> None: + left = { + "server": { + "name": "example", + "version": "1", + "remotes": [{"url": "https://example.invalid"}], + }, + OFFICIAL_META_KEY: {"status": "active"}, + } + right = { + OFFICIAL_META_KEY: {"status": "active"}, + "server": { + "remotes": [{"url": "https://example.invalid"}], + "version": "1", + "name": "example", + }, + } + assert record_hash(left) == record_hash(right) + + +def test_record_hash_includes_official_metadata() -> None: + server = { + "name": "example", + "version": "1", + "remotes": [{"url": "https://example.invalid"}], + } + active = {"server": server, OFFICIAL_META_KEY: {"status": "active"}} + deprecated = {"server": server, OFFICIAL_META_KEY: {"status": "deprecated"}} + assert record_hash(active) != record_hash(deprecated) + + +def test_posture_findings_cover_registry_boundaries() -> None: + findings = [ + finding + for snapshot in normalize_payload(payload(), source="fixture") + for finding in posture_findings(snapshot) + ] + ids = {finding["id"] for finding in findings} + assert {"MCP-REPOSITORY", "MCP-OFFICIAL-STATUS", "MCP-PLAIN-HTTP"} <= ids + + +def test_posture_flags_unrecognized_status_instead_of_failing() -> None: + snapshot = normalize_payload( + one_server(pinned_server(), official={"status": "suspended"}), source="fixture" + )[0] + findings = posture_findings(snapshot) + assert [finding["id"] for finding in findings] == ["MCP-OFFICIAL-STATUS"] + assert findings[0]["message"] == "Official status is suspended" + assert findings[0]["risk_score"] > 0 + + +def test_posture_flags_unpinned_package_and_missing_hash() -> None: + server = pinned_server(packages=[pinned_package(version="latest")]) + del server["packages"][0]["fileSha256"] + snapshot = normalize_payload( + one_server(server, official={"status": "active"}), source="fixture" + )[0] + findings = posture_findings(snapshot) + by_id = {finding["id"]: finding for finding in findings} + assert by_id["MCP-PACKAGE-VERSION"]["evidence"] == "registry_assertion" + assert by_id["MCP-PACKAGE-SHA256"]["evidence"] == "unavailable" + + +def test_posture_flags_missing_status_as_unavailable() -> None: + snapshot = normalize_payload(one_server(pinned_server()), source="fixture")[0] + findings = posture_findings(snapshot) + assert [finding["id"] for finding in findings] == ["MCP-OFFICIAL-STATUS"] + assert findings[0]["evidence"] == "unavailable" + assert findings[0]["risk_score"] == 0 + + +def test_negative_space_pinned_active_server_has_no_findings() -> None: + snapshot = normalize_payload( + one_server(pinned_server(), official={"status": "active"}), source="fixture" + )[0] + assert posture_findings(snapshot) == [] + + +def test_negative_space_absent_optional_facts_are_unavailable() -> None: + snapshot = normalize_payload( + {"servers": [{"server": {"name": "unknown/example"}}]}, source="fixture" + )[0] + findings = posture_findings(snapshot) + assert findings + assert all(finding["evidence"] == "unavailable" for finding in findings) + assert all(finding["risk_score"] == 0 for finding in findings) + + +@pytest.mark.parametrize( + "version, file_sha256", + [("latest", "not-a-sha256"), ("", "")], +) +def test_negative_space_asserted_bad_version_and_hash_are_flagged( + version: str, file_sha256: str +) -> None: + server = pinned_server(packages=[pinned_package(version=version, fileSha256=file_sha256)]) + snapshot = normalize_payload( + one_server(server, official={"status": "active"}), source="fixture" + )[0] + findings = posture_findings(snapshot) + ids = {finding["id"] for finding in findings} + assert {"MCP-PACKAGE-VERSION", "MCP-PACKAGE-SHA256"} <= ids + assert {finding["evidence"] for finding in findings} == {"registry_assertion"} + + +@pytest.mark.parametrize( + "version", + ["latest", "LATEST", "next", "beta", "1", "1.2", "1.x", "1.0.0||2.0.0"], +) +def test_negative_space_mutable_version_tags_are_flagged(version: str) -> None: + server = pinned_server(packages=[pinned_package(version=version)]) + snapshot = normalize_payload( + one_server(server, official={"status": "active"}), source="fixture" + )[0] + ids = {finding["id"] for finding in posture_findings(snapshot)} + assert "MCP-PACKAGE-VERSION" in ids + + +@pytest.mark.parametrize( + "registry_type, version", + [("npm", "1.0.0-experimental"), ("npm", "1.0.0+linux-x64"), ("oci", "1.0.0-linux-x64")], +) +def test_negative_space_exact_versions_with_suffixes_are_pinned( + registry_type: str, version: str +) -> None: + server = pinned_server(packages=[pinned_package(registryType=registry_type, version=version)]) + snapshot = normalize_payload( + one_server(server, official={"status": "active"}), source="fixture" + )[0] + assert posture_findings(snapshot) == [] + + +def test_negative_space_empty_repository_url_is_unavailable() -> None: + server = pinned_server(repository={"url": "", "source": "github"}) + snapshot = normalize_payload( + one_server(server, official={"status": "active"}), source="fixture" + )[0] + findings = posture_findings(snapshot) + assert [finding["id"] for finding in findings] == ["MCP-REPOSITORY"] + assert findings[0]["evidence"] == "unavailable" + + +def test_error_on_malformed_payload() -> None: + with pytest.raises(ValueError, match="MCP Registry"): + scan_registry(str(FIXTURES / "malformed.json")) + + +def test_error_on_missing_servers_list() -> None: + with pytest.raises(ValueError, match="servers list"): + normalize_payload({}, source="fixture") + + +@pytest.mark.parametrize( + "field_name, value", [("packages", {}), ("packages", None), ("remotes", {})] +) +def test_error_on_invalid_collection_shapes(field_name: str, value: object) -> None: + with pytest.raises(ValueError, match="invalid .* collection"): + normalize_payload( + {"servers": [{"server": {"name": "broken/example", field_name: value}}]}, + source="fixture", + ) + + +def test_error_on_invalid_repository_shape() -> None: + with pytest.raises(ValueError, match="invalid repository object"): + normalize_payload( + {"servers": [{"server": {"name": "broken/example", "repository": []}}]}, + source="fixture", + ) + + +def test_error_on_repeated_next_cursor(monkeypatch: pytest.MonkeyPatch) -> None: + page = {"servers": [{"server": {"name": "page/one"}}], "metadata": {"nextCursor": "cursor-1"}} + + def fake_get(url: str, *, params: dict[str, str] | None = None, timeout: int) -> _Response: + return _Response(page) + + monkeypatch.setattr("skillspector.mcp_registry.httpx.get", fake_get) + + with pytest.raises(ValueError, match="repeated pagination cursor"): + scan_registry(REGISTRY_URL) + + +def test_error_on_http_failure(monkeypatch: pytest.MonkeyPatch) -> None: + def fake_get(url: str, *, params: dict[str, str] | None = None, timeout: int) -> _Response: + raise httpx.HTTPError("network down") + + monkeypatch.setattr("skillspector.mcp_registry.httpx.get", fake_get) + + with pytest.raises(ValueError, match="MCP Registry source failed"): + scan_registry(REGISTRY_URL) + + +def test_scan_registry_scans_partial_paginated_capture(tmp_path: Path) -> None: + capture = tmp_path / "registry.json" + capture.write_text( + json.dumps( + { + "servers": [{"server": {"name": "page/one"}}], + "metadata": {"nextCursor": "page-2"}, + } + ), + encoding="utf-8", + ) + report = scan_registry(str(capture)) + assert report["server_count"] == 1 + assert report["servers"][0]["snapshot"]["name"] == "page/one" + + +def test_scan_registry_aggregates_risk_score() -> None: + report = scan_registry(str(FIXTURES / "mcp_registry.json")) + assert report["risk_score"] == 45 + assert report["max_risk_score"] == 25