diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 3b15b4e44..462f5d59e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -102,6 +102,8 @@ jobs: with: node-version: ${{ env.NODE_VERSION }} - run: node --test tests/opencode/skillspector_scan_lib.test.ts + - name: Run Pi extension security tests + run: node --test tests/unit/test_pi_extension.mjs docker-smoke: needs: changes diff --git a/extensions/skillspector.ts b/extensions/skillspector.ts index 6a9e27341..bec77df11 100644 --- a/extensions/skillspector.ts +++ b/extensions/skillspector.ts @@ -1,8 +1,9 @@ import type { ExtensionAPI } from "@earendil-works/pi-coding-agent"; import { StringEnum } from "@earendil-works/pi-ai"; import { Type, type Static } from "typebox"; -import { existsSync } from "node:fs"; -import { dirname, isAbsolute, resolve } from "node:path"; +import { chmodSync, constants, copyFileSync, existsSync, lstatSync, mkdtempSync, realpathSync, renameSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { basename, dirname, isAbsolute, join, relative, resolve, sep } from "node:path"; import { fileURLToPath } from "node:url"; const scanSchema = Type.Object({ @@ -12,7 +13,7 @@ const scanSchema = Type.Object({ description: "SkillSpector output format. Defaults to terminal.", }), ), - output: Type.Optional(Type.String({ description: "Optional report output path." })), + output: Type.Optional(Type.String({ description: "Optional report output path within the current workspace." })), noLlm: Type.Optional(Type.Boolean({ description: "Skip LLM analysis. Defaults to true." })), provider: Type.Optional( StringEnum(["openai", "anthropic", "anthropic_proxy", "nv_build", "nv_inference"] as const, { @@ -57,20 +58,57 @@ function packageRoot(): string { } function findSkillSpectorBin(): string { - if (process.env.SKILLSPECTOR_BIN) return process.env.SKILLSPECTOR_BIN; - const localBin = resolve(packageRoot(), ".venv/bin/skillspector"); - if (existsSync(localBin)) return localBin; - return "skillspector"; + const bundledBin = process.platform === "win32" ? ".venv/Scripts/skillspector.exe" : ".venv/bin/skillspector"; + const bin = process.env.SKILLSPECTOR_BIN ?? resolve(packageRoot(), bundledBin); + if (!isAbsolute(bin) || !existsSync(bin)) { + throw new Error("Install SkillSpector in the extension's .venv or set SKILLSPECTOR_BIN to an absolute executable path."); + } + return bin; } -function buildScanArgs(params: SkillSpectorScanParams, cwd: string): string[] { +function isWithin(root: string, path: string): boolean { + const rel = relative(root, path); + return rel !== ".." && !rel.startsWith(`..${sep}`) && !isAbsolute(rel); +} + +function reportOutputPath(cwd: string, value?: string): string | undefined { + if (!value) return undefined; + const output = resolve(cwd, value); + if (output === resolve(cwd) || !isWithin(resolve(cwd), output)) { + throw new Error("Report output must be a file within the current workspace."); + } + const parent = realpathSync(dirname(output)); + if (!isWithin(realpathSync(cwd), parent)) { + throw new Error("Report output must be a file within the current workspace."); + } + const destination = join(parent, basename(output)); + const existing = lstatSync(destination, { throwIfNoEntry: false }); + if (existing && !existing.isFile()) { + throw new Error("Report output must be a regular file, not a symlink or directory."); + } + return destination; +} + +function publishReport(source: string, destination: string): void { + const staging = mkdtempSync(join(dirname(destination), ".skillspector-report-")); + try { + const report = join(staging, "report"); + copyFileSync(source, report, constants.COPYFILE_EXCL); + chmodSync(report, 0o600); + // Replacing the directory entry never follows an existing hard link. + renameSync(report, destination); + } finally { + rmSync(staging, { recursive: true, force: true }); + } +} + +function buildScanArgs(params: SkillSpectorScanParams, cwd: string, output?: string): string[] { const args = ["scan", resolveMaybePath(cwd, params.target) ?? params.target]; args.push("--format", params.format ?? "terminal"); const noLlm = params.noLlm ?? true; if (noLlm) args.push("--no-llm"); - const output = resolveMaybePath(cwd, params.output); if (output) args.push("--output", output); const yaraRulesDir = resolveMaybePath(cwd, params.yaraRulesDir); @@ -93,50 +131,65 @@ export default function (pi: ExtensionAPI) { parameters: scanSchema, async execute(_toolCallId, params, signal, onUpdate, ctx) { const bin = findSkillSpectorBin(); - const args = buildScanArgs(params, ctx.cwd); - const env: Record = {}; - - if (params.provider) env.SKILLSPECTOR_PROVIDER = params.provider; - if (params.model) env.SKILLSPECTOR_MODEL = params.model; - - onUpdate?.({ content: [{ type: "text", text: `Running ${bin} ${args.slice(0, 2).join(" ")} ...` }] }); - - const result = await pi.exec(bin, args, { - cwd: ctx.cwd, - env, - signal, - timeout: 120000, - }); - - const stdout = truncateText(redactSecrets(result.stdout ?? "")); - const stderr = truncateText(redactSecrets(result.stderr ?? ""), 6000); - - if (result.code !== 0) { - throw new Error(`SkillSpector failed with exit code ${result.code}.\n${stderr.text}`); + const outputPath = reportOutputPath(ctx.cwd, params.output); + const reportDir = outputPath ? mkdtempSync(join(tmpdir(), "skillspector-report-")) : undefined; + const reportPath = reportDir ? join(reportDir, "report") : undefined; + try { + const args = buildScanArgs(params, ctx.cwd, reportPath); + const env: Record = {}; + + if (params.provider) env.SKILLSPECTOR_PROVIDER = params.provider; + if (params.model) env.SKILLSPECTOR_MODEL = params.model; + + onUpdate?.({ content: [{ type: "text", text: `Running ${bin} ${args.slice(0, 2).join(" ")} ...` }] }); + + const result = await pi.exec(bin, args, { + cwd: ctx.cwd, + env, + signal, + timeout: 120000, + }); + + const stdout = truncateText(redactSecrets(result.stdout ?? "")); + const stderr = truncateText(redactSecrets(result.stderr ?? ""), 6000); + + // Exit 2 can follow a diagnostic report; failures before reporting produce no file. + const failureReport = result.code === 2 && reportPath + ? lstatSync(reportPath, { throwIfNoEntry: false }) : undefined; + const hasFailureReport = failureReport?.isFile() && failureReport.size > 0; + if ((result.code === 0 || result.code === 1 || hasFailureReport) && outputPath && reportPath) { + if (reportOutputPath(ctx.cwd, params.output) !== outputPath) { + throw new Error("Report output directory changed during the scan."); + } + publishReport(reportPath, outputPath); + } + if (result.code !== 0) { + throw new Error(`SkillSpector failed with exit code ${result.code}.\n${stderr.text}`); + } + const lines = [ + `SkillSpector scan complete: ${params.target}`, + `format: ${params.format ?? "terminal"}`, + `noLlm: ${params.noLlm ?? true}`, + ]; + if (outputPath) lines.push(`output: ${outputPath}`); + if (stdout.truncated) lines.push("stdout truncated: true"); + if (stderr.text.trim()) lines.push(`stderr:\n${stderr.text}`); + if (stdout.text.trim()) lines.push(`stdout:\n${stdout.text}`); + + return { + content: [{ type: "text", text: lines.join("\n") }], + details: { + code: result.code, + command: bin, + args, + outputPath, + stdoutTruncated: stdout.truncated, + stderrTruncated: stderr.truncated, + }, + }; + } finally { + if (reportDir) rmSync(reportDir, { recursive: true, force: true }); } - - const outputPath = resolveMaybePath(ctx.cwd, params.output); - const lines = [ - `SkillSpector scan complete: ${params.target}`, - `format: ${params.format ?? "terminal"}`, - `noLlm: ${params.noLlm ?? true}`, - ]; - if (outputPath) lines.push(`output: ${outputPath}`); - if (stdout.truncated) lines.push("stdout truncated: true"); - if (stderr.text.trim()) lines.push(`stderr:\n${stderr.text}`); - if (stdout.text.trim()) lines.push(`stdout:\n${stdout.text}`); - - return { - content: [{ type: "text", text: lines.join("\n") }], - details: { - code: result.code, - command: bin, - args, - outputPath, - stdoutTruncated: stdout.truncated, - stderrTruncated: stderr.truncated, - }, - }; }, }); } diff --git a/src/skillspector/mcp_registry.py b/src/skillspector/mcp_registry.py index 5678cda5b..9733f4af8 100644 --- a/src/skillspector/mcp_registry.py +++ b/src/skillspector/mcp_registry.py @@ -7,17 +7,22 @@ import hashlib import json +import os import re from dataclasses import asdict, dataclass from datetime import UTC, datetime from itertools import chain from pathlib import Path +from stat import S_ISREG from typing import Any, TypedDict import httpx REGISTRY_URL = "https://registry.modelcontextprotocol.io/v0/servers" OFFICIAL_META_KEY = "io.modelcontextprotocol.registry/official" +MAX_REGISTRY_BYTES = 16 * 1024 * 1024 +MAX_REGISTRY_DEPTH = 64 +MAX_REGISTRY_RECORDS = 10_000 FILE_SHA256_RE = re.compile(r"^[a-f0-9]{64}$") MUTABLE_VERSION_TAGS = frozenset( { @@ -224,6 +229,24 @@ def normalize_server( 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") + # Check the full expansion before constructing any snapshots or findings. + records = len(payload["servers"]) + if records <= MAX_REGISTRY_RECORDS: + for entry in payload["servers"]: + record = entry.get("server") if isinstance(entry, dict) else None + if isinstance(record, dict): + records += sum( + len(record[field]) + for field in ("packages", "remotes") + if isinstance(record.get(field), list) + ) + if records > MAX_REGISTRY_RECORDS: + break + if records > MAX_REGISTRY_RECORDS: + raise ValueError( + f"MCP Registry payload from {source} exceeds {MAX_REGISTRY_RECORDS} records " + "(servers, packages, and remotes)" + ) scanned_at = datetime.now(UTC).isoformat() return [ normalize_server(entry, source=source, scanned_at=scanned_at) @@ -344,14 +367,47 @@ def _dict_payload(payload: object, *, source: str) -> dict[str, Any]: return payload +def _load_local_registry(path: Path) -> dict[str, Any]: + with open( + path, + "rb", + opener=lambda path, flags: os.open(path, flags | getattr(os, "O_NONBLOCK", 0)), + ) as source_file: + if not S_ISREG(os.fstat(source_file.fileno()).st_mode): + raise ValueError(f"Registry input must be a regular file: {path}") + raw = source_file.read(MAX_REGISTRY_BYTES + 1) + if len(raw) > MAX_REGISTRY_BYTES: + raise ValueError(f"input exceeds {MAX_REGISTRY_BYTES} bytes") + text = raw.decode("utf-8") + # The JSON decoder owns syntax validation; only bound nesting here, while + # ignoring delimiters in strings, before its recursive parser runs. + depth = 0 + in_string = False + escaped = False + for char in text: + if in_string: + if escaped: + escaped = False + elif char == "\\": + escaped = True + elif char == '"': + in_string = False + elif char == '"': + in_string = True + elif char in "[{": + depth += 1 + if depth > MAX_REGISTRY_DEPTH: + raise ValueError(f"JSON nesting exceeds {MAX_REGISTRY_DEPTH} levels") + elif char in "]}": + depth -= 1 + return _dict_payload(json.loads(text), source=str(path)) + + 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, - ) + return _load_local_registry(Path(input_path)) if input_path.startswith(("http://", "https://")): if input_path != REGISTRY_URL: raise ValueError( @@ -372,7 +428,7 @@ def _load_payload(input_path: str) -> dict[str, Any]: # 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: + except (OSError, httpx.HTTPError, ValueError, RecursionError) 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 diff --git a/src/skillspector/providers/_agent_cli.py b/src/skillspector/providers/_agent_cli.py index 58425f1f0..b085fa8b7 100644 --- a/src/skillspector/providers/_agent_cli.py +++ b/src/skillspector/providers/_agent_cli.py @@ -79,7 +79,10 @@ # Includes API keys, cloud creds, SSH agent, and SkillSpector's own keys. _SECRET_ENV_PREFIXES: tuple[str, ...] = ( "ANTHROPIC_API_KEY", + "ANTHROPIC_PROXY_API_KEY", "OPENAI_API_KEY", + "SKILLSPECTOR_API_KEYS", + "SKILLSPECTOR_COMPAT_API_KEY", "NVIDIA_INFERENCE_KEY", "NVIDIA_INFERENCE_METADATA_KEY", "AWS_", diff --git a/src/skillspector/suppression.py b/src/skillspector/suppression.py index d8b5ee23f..61809eb35 100644 --- a/src/skillspector/suppression.py +++ b/src/skillspector/suppression.py @@ -58,11 +58,13 @@ import fnmatch import hashlib import json +import os import posixpath import re from collections.abc import Mapping from dataclasses import dataclass, field from pathlib import Path +from stat import S_ISREG from typing import Any import yaml @@ -73,11 +75,72 @@ logger = get_logger(__name__) BASELINE_VERSION = 2 +MAX_BASELINE_BYTES = 2 * 1024 * 1024 +MAX_BASELINE_NODES = 100_000 +MAX_BASELINE_DEPTH = 64 +MAX_BASELINE_RECORDS = 10_000 +MAX_BASELINE_SCALAR_CHARS = 64 * 1024 _FINGERPRINT_SCHEMA = "skillspector-finding-fingerprint-v2" _FINGERPRINT_RE = re.compile(r"sha256:[0-9a-f]{64}\Z") _SOURCE_IDENTITY_RE = re.compile(r"external/[0-9a-f]{64}\Z") +class _BoundedBaselineLoader(yaml.SafeLoader): + """Bound composition and alias expansion before YAML object construction.""" + + def __init__(self, stream: str) -> None: + super().__init__(stream) + self._nodes = 0 + self._depth = 0 + + def compose_node(self, parent: object, index: object) -> yaml.Node: + self._nodes += 1 + if self._nodes > MAX_BASELINE_NODES: + raise ValueError("baseline exceeds YAML node limit") + if self._depth >= MAX_BASELINE_DEPTH: + raise ValueError("baseline exceeds YAML depth limit") + self._depth += 1 + try: + return super().compose_node(parent, index) + finally: + self._depth -= 1 + + def construct_document(self, node: yaml.Node) -> Any: + nodes = 0 + characters = 0 + active: set[int] = set() + + def visit(item: yaml.Node, depth: int) -> None: + nonlocal nodes, characters + nodes += 1 + if nodes > MAX_BASELINE_NODES: + raise ValueError("baseline exceeds expanded YAML node limit") + if depth > MAX_BASELINE_DEPTH: + raise ValueError("baseline exceeds expanded YAML depth limit") + if id(item) in active: + raise ValueError("baseline contains cyclic YAML aliases") + if isinstance(item, yaml.ScalarNode): + if len(item.value) > MAX_BASELINE_SCALAR_CHARS: + raise ValueError("baseline exceeds YAML scalar character limit") + characters += len(item.value) + if characters > MAX_BASELINE_BYTES: + raise ValueError("baseline exceeds expanded YAML character limit") + return + active.add(id(item)) + children = ( + (child for pair in item.value for child in pair) + if isinstance(item, yaml.MappingNode) + else iter(item.value) + ) + for child in children: + visit(child, depth + 1) + active.remove(id(item)) + + # Revisit each alias: counting only unique nodes misses merge amplification. + visit(node, 1) + return super().construct_document(node) + + def _has_exact_source_provenance(finding: Finding) -> bool: """Return whether immutable transitive provenance has canonical form.""" return bool( @@ -308,7 +371,16 @@ def baseline_from_dict(data: dict[str, Any]) -> Baseline: raise ValueError(f"baseline must be a mapping (got {type(data).__name__})") version = data.get("version") - raw_fingerprints = data.get("fingerprints") or [] + raw_rules = data.get("rules") + raw_fingerprints = data.get("fingerprints") + if raw_rules is None: + raw_rules = [] + if raw_fingerprints is None: + raw_fingerprints = [] + if not isinstance(raw_rules, list) or not isinstance(raw_fingerprints, list): + raise ValueError("baseline rules and fingerprints must be lists") + if len(raw_rules) + len(raw_fingerprints) > MAX_BASELINE_RECORDS: + raise ValueError(f"baseline exceeds record limit ({MAX_BASELINE_RECORDS})") is_legacy_rule_only = version in (None, 1) and not raw_fingerprints if version != BASELINE_VERSION and not is_legacy_rule_only: migration = ( @@ -328,7 +400,7 @@ def baseline_from_dict(data: dict[str, Any]) -> Baseline: ) rules: list[SuppressionRule] = [] - for raw in data.get("rules") or []: + for raw in raw_rules: if not isinstance(raw, dict): raise ValueError(f"each baseline rule must be a mapping, got: {raw!r}") reason = raw.get("reason", "") @@ -382,10 +454,19 @@ def load_baseline(path: str | Path) -> Baseline: p = Path(path) if not p.exists(): raise FileNotFoundError(f"Baseline file not found: {p}") - text = p.read_text(encoding="utf-8") + with open( + p, + "rb", + opener=lambda path, flags: os.open(path, flags | getattr(os, "O_NONBLOCK", 0)), + ) as source: + if not S_ISREG(os.fstat(source.fileno()).st_mode): + raise ValueError(f"Baseline must be a regular file: {p}") + content = source.read(MAX_BASELINE_BYTES + 1) + if len(content) > MAX_BASELINE_BYTES: + raise ValueError(f"Baseline file exceeds byte limit ({MAX_BASELINE_BYTES}): {p}") try: - # yaml.safe_load parses JSON too, so a single path handles both formats. - data = yaml.safe_load(text) or {} + # SafeLoader parses JSON too, so a single path handles both formats. + data = yaml.load(content.decode("utf-8"), Loader=_BoundedBaselineLoader) or {} except yaml.YAMLError as e: # pragma: no cover - error path raise ValueError(f"Could not parse baseline file {p}: {e}") from e return baseline_from_dict(data) diff --git a/tests/unit/test_agent_cli.py b/tests/unit/test_agent_cli.py index 8535bd808..4e1c73cb4 100644 --- a/tests/unit/test_agent_cli.py +++ b/tests/unit/test_agent_cli.py @@ -309,6 +309,38 @@ def test_preserves_safe_vars(self, monkeypatch: pytest.MonkeyPatch) -> None: assert "PATH" in env assert "HOME" in env + @pytest.mark.parametrize("binary_name", ["claude", "codex", "gemini"]) + @pytest.mark.parametrize("lowercase", [False, True]) + def test_project_credentials_never_reach_child( + self, monkeypatch: pytest.MonkeyPatch, binary_name: str, lowercase: bool + ) -> None: + secrets = [ + "SKILLSPECTOR_API_KEYS", + "ANTHROPIC_PROXY_API_KEY", + "SKILLSPECTOR_COMPAT_API_KEY", + ] + if lowercase: + secrets = [key.lower() for key in secrets] + for key in secrets: + monkeypatch.setenv(key, "synthetic-credential") + monkeypatch.setenv("SKILLSPECTOR_MODEL", "test-model") + monkeypatch.setenv("SKILLSPECTOR_COMPAT_BASE_URL", "https://example.invalid") + output = _GOOD_CODEX_JSONL if binary_name == "codex" else _GOOD_CLAUDE_OUTPUT + with ( + patch( + "skillspector.providers._agent_cli.find_binary", return_value="/usr/bin/mock-cli" + ), + patch("skillspector.providers._agent_cli.subprocess.Popen") as popen, + ): + popen.return_value = _make_ok_process(output.encode()) + run_agent_cli(binary_name, PROMPT, model="") + child_env = popen.call_args.kwargs["env"] + assert not set(secrets).intersection(child_env) + assert child_env["SKILLSPECTOR_MODEL"] == "test-model" + assert child_env["SKILLSPECTOR_COMPAT_BASE_URL"] == "https://example.invalid" + for key in secrets: + assert _agent_cli.os.environ[key] == "synthetic-credential" + # --------------------------------------------------------------------------- # _parse_claude_output diff --git a/tests/unit/test_cli.py b/tests/unit/test_cli.py index 3208b9872..a868ff7e8 100644 --- a/tests/unit/test_cli.py +++ b/tests/unit/test_cli.py @@ -937,6 +937,38 @@ def test_cli_scan_missing_baseline_exits_2(tmp_path: Path) -> None: assert "baseline" in result.output.lower() +@pytest.mark.parametrize("mode", ["registry", "explicit_baseline", "shipped_baseline"]) +def test_cli_oversized_input_exits_2_without_replacing_report( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, mode: str +) -> None: + from skillspector import mcp_registry, suppression + + skill = tmp_path / "skill" + skill.mkdir() + (skill / "SKILL.md").write_text("# A simple skill\n", encoding="utf-8") + output = tmp_path / "report.json" + output.write_text("previous report", encoding="utf-8") + if mode == "registry": + source = tmp_path / "registry.json" + source.write_text('{"servers": []}', encoding="utf-8") + monkeypatch.setattr(mcp_registry, "MAX_REGISTRY_BYTES", 8, raising=False) + args = ["scan", str(source), "--mcp-registry"] + else: + source = skill / ".skillspector-baseline.yaml" + source.write_text("version: 2\nrules: []\n", encoding="utf-8") + monkeypatch.setattr(suppression, "MAX_BASELINE_BYTES", 8, raising=False) + args = ["scan", str(skill), "--no-llm"] + args += ( + ["--baseline", str(source)] + if mode == "explicit_baseline" + else ["--use-shipped-baseline"] + ) + result = runner.invoke(app, [*args, "--format", "json", "--output", str(output)]) + assert result.exit_code == 2 + assert "exceeds" in result.output + assert output.read_text(encoding="utf-8") == "previous report" + + def test_cli_baseline_generate_then_scan_round_trip(tmp_path: Path) -> None: """`baseline` writes a file; scanning with it suppresses those findings.""" skill = tmp_path / "skill" diff --git a/tests/unit/test_mcp_registry.py b/tests/unit/test_mcp_registry.py index 8dac6aa12..b1f060452 100644 --- a/tests/unit/test_mcp_registry.py +++ b/tests/unit/test_mcp_registry.py @@ -4,12 +4,17 @@ """Tests for the MCP Registry owner and posture checks.""" import json +import os +import subprocess +import sys +from io import BufferedReader from pathlib import Path from typing import Any import httpx import pytest +from skillspector import mcp_registry from skillspector.mcp_registry import ( OFFICIAL_META_KEY, REGISTRY_URL, @@ -419,3 +424,147 @@ 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 + + +def test_local_registry_bounds_read_before_parsing( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + capture = tmp_path / "registry.json" + capture.write_bytes(b" " * 64) + monkeypatch.setattr(mcp_registry, "MAX_REGISTRY_BYTES", 32, raising=False) + read_sizes: list[int] = [] + + class Source(BufferedReader): + def read(self, size: int = -1) -> bytes: + read_sizes.append(size) + return super().read(size) + + def unexpected_parse(*args: Any, **kwargs: Any) -> None: + pytest.fail("oversized JSON must be rejected before parsing") + + monkeypatch.setattr( + mcp_registry, + "open", + lambda *args, **kwargs: Source(open(*args, **kwargs, buffering=0)), + raising=False, + ) + monkeypatch.setattr(mcp_registry.json, "loads", unexpected_parse) + with pytest.raises(ValueError, match="MCP Registry source failed.*exceeds 32 bytes"): + scan_registry(str(capture)) + assert read_sizes == [33] + + +@pytest.mark.skipif(not hasattr(os, "mkfifo"), reason="requires POSIX FIFO support") +def test_local_registry_rejects_fifo_swap_without_blocking(tmp_path: Path) -> None: + capture = tmp_path / "registry.json" + capture.write_text('{"servers": []}', encoding="utf-8") + # Keep a regressed blocking open contained in a child with a hard timeout. + result = subprocess.run( + [ + sys.executable, + "-c", + """ +import os +import sys +from pathlib import Path +from unittest.mock import patch + +from skillspector.mcp_registry import scan_registry + +capture = Path(sys.argv[1]) +original_is_file = Path.is_file + +def swap_after_check(path): + is_file = original_is_file(path) + if path == capture and is_file: + path.unlink() + os.mkfifo(path) + return is_file + +with patch.object(Path, "is_file", swap_after_check): + try: + scan_registry(str(capture)) + except ValueError as exc: + assert "MCP Registry source failed" in str(exc), str(exc) + assert "regular file" in str(exc), str(exc) + else: + raise AssertionError("swapped FIFO must be rejected") +""", + str(capture), + ], + capture_output=True, + text=True, + timeout=5, + check=False, + ) + assert result.returncode == 0, result.stdout + result.stderr + + +def test_local_registry_accepts_exact_byte_limit( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + capture = tmp_path / "registry.json" + raw = json.dumps(one_server({"name": "example/ΓΌ"}), ensure_ascii=False).encode("utf-8") + capture.write_bytes(raw) + monkeypatch.setattr(mcp_registry, "MAX_REGISTRY_BYTES", len(raw), raising=False) + assert scan_registry(str(capture))["server_count"] == 1 + + +def test_local_registry_bounds_depth_before_parsing( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + capture = tmp_path / "registry.json" + capture.write_text('{"servers": [], "extra": [[[]]]}', encoding="utf-8") + monkeypatch.setattr(mcp_registry, "MAX_REGISTRY_DEPTH", 3, raising=False) + + def unexpected_parse(*args: Any, **kwargs: Any) -> None: + pytest.fail("deep JSON must be rejected before parsing") + + monkeypatch.setattr(mcp_registry.json, "loads", unexpected_parse) + with pytest.raises(ValueError, match="MCP Registry source failed.*nesting exceeds 3 levels"): + scan_registry(str(capture)) + + +def test_local_registry_depth_ignores_escaped_string_contents( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + capture = tmp_path / "registry.json" + server = {"name": "example/safe", "description": '[[[{{{\\"\\\\'} + capture.write_text(json.dumps(one_server(server)), encoding="utf-8") + monkeypatch.setattr(mcp_registry, "MAX_REGISTRY_DEPTH", 4, raising=False) + assert scan_registry(str(capture))["snapshots"][0]["description"] == server["description"] + + +@pytest.mark.parametrize("field", ["servers", "packages", "remotes"]) +def test_registry_bounds_record_expansion_before_normalization( + field: str, monkeypatch: pytest.MonkeyPatch +) -> None: + data = one_server({"name": "example/safe"}) + if field == "servers": + data["servers"] *= 4 + else: + data["servers"][0]["server"][field] = [{}, {}, {}] + monkeypatch.setattr(mcp_registry, "MAX_REGISTRY_RECORDS", 3, raising=False) + + def unexpected_normalize(*args: Any, **kwargs: Any) -> None: + pytest.fail("record budget must be checked before constructing snapshots") + + monkeypatch.setattr(mcp_registry, "normalize_server", unexpected_normalize) + with pytest.raises(ValueError, match="exceeds 3 records"): + normalize_payload(data, source="fixture") + + +def test_registry_record_budget_is_aggregate_and_accepts_exact_limit( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + capture = tmp_path / "registry.json" + data = one_server(pinned_server()) + data["servers"].append({"server": pinned_server(name="example/second")}) + capture.write_text(json.dumps(data), encoding="utf-8") + monkeypatch.setattr(mcp_registry, "MAX_REGISTRY_RECORDS", 6, raising=False) + report = scan_registry(str(capture)) + assert report["server_count"] == 2 + assert len(report["snapshots"]) == len(report["servers"]) == 2 + monkeypatch.setattr(mcp_registry, "MAX_REGISTRY_RECORDS", 5) + with pytest.raises(ValueError, match="exceeds 5 records"): + scan_registry(str(capture)) diff --git a/tests/unit/test_pi_extension.mjs b/tests/unit/test_pi_extension.mjs new file mode 100644 index 000000000..68ad607f2 --- /dev/null +++ b/tests/unit/test_pi_extension.mjs @@ -0,0 +1,237 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +// Run with Node >= 22.18: node --test tests/unit/test_pi_extension.mjs +// The CLI and schema-only peer dependencies are mocked; no subprocess is started. +import assert from "node:assert/strict"; +import { copyFileSync, existsSync, linkSync, mkdirSync, mkdtempSync, readFileSync, + readdirSync, realpathSync, renameSync, rmSync, statSync, symlinkSync, writeFileSync } from "node:fs"; +import { registerHooks } from "node:module"; +import { tmpdir } from "node:os"; +import { dirname, join, resolve } from "node:path"; +import test from "node:test"; +import { pathToFileURL } from "node:url"; + +registerHooks({ + resolve(specifier, context, nextResolve) { + const stub = specifier === "@earendil-works/pi-ai" + ? "export const StringEnum = () => ({});" + : specifier === "typebox" + ? "export const Type = {Object: x => x, Optional: x => x, String: () => ({}), Boolean: () => ({})};" + : undefined; + return stub + ? { url: `data:text/javascript,${encodeURIComponent(stub)}`, shortCircuit: true } + : nextResolve(specifier, context); + }, +}); + +async function setup(t, exec) { + const root = realpathSync(mkdtempSync(join(tmpdir(), "skillspector-pi-test-"))); + const workspace = join(root, "workspace"); + const install = join(root, "install"); + const bin = join(install, process.platform === "win32" ? ".venv/Scripts/skillspector.exe" : ".venv/bin/skillspector"); + const source = process.env.SKILLSPECTOR_EXTENSION_SOURCE + ?? new URL("../../extensions/skillspector.ts", import.meta.url); + mkdirSync(workspace); + mkdirSync(join(install, "extensions"), { recursive: true }); + mkdirSync(dirname(bin), { recursive: true }); + writeFileSync(bin, "unused mocked executable"); + writeFileSync(join(install, "package.json"), '{"type":"module"}'); + copyFileSync(source, join(install, "extensions/skillspector.ts")); + const originalBin = process.env.SKILLSPECTOR_BIN; + delete process.env.SKILLSPECTOR_BIN; + t.after(() => { + if (originalBin === undefined) delete process.env.SKILLSPECTOR_BIN; + else process.env.SKILLSPECTOR_BIN = originalBin; + rmSync(root, { recursive: true, force: true }); + }); + const { default: register } = await import(pathToFileURL(join(install, "extensions/skillspector.ts"))); + let tool; + const calls = []; + register({ + registerTool(registered) { tool = registered; }, + async exec(command, args, options) { + const outputIndex = args.indexOf("--output"); + const output = outputIndex < 0 ? undefined : resolve(options.cwd, args[outputIndex + 1]); + calls.push({ command, args, options, output }); + if (output) writeFileSync(output, "new report"); + return exec?.({ command, args, options, output, root, workspace }) + ?? { code: 0, stdout: "complete", stderr: "" }; + }, + }); + return { + root, workspace, bin, calls, + scan: (params = {}) => tool.execute("scan", { target: "./SKILL.md", ...params }, undefined, undefined, { cwd: workspace }), + }; +} + +test("uses installed absolute executable and preserves scan arguments without output", async (t) => { + const ctx = await setup(t); + await ctx.scan({ provider: "anthropic", model: "synthetic-model", verbose: true }); + assert.equal(ctx.calls[0].command, ctx.bin); + assert.deepEqual(ctx.calls[0].args, ["scan", "./SKILL.md", "--format", "terminal", "--no-llm", "--verbose"]); + assert.deepEqual(ctx.calls[0].options.env, { SKILLSPECTOR_PROVIDER: "anthropic", SKILLSPECTOR_MODEL: "synthetic-model" }); + assert.equal(ctx.calls[0].options.cwd, ctx.workspace); +}); + +test("finds the Windows virtualenv executable without a PATH fallback", async (t) => { + const ctx = await setup(t); + rmSync(ctx.bin); + const windowsBin = join(ctx.root, "install/.venv/Scripts/skillspector.exe"); + mkdirSync(dirname(windowsBin), { recursive: true }); + writeFileSync(windowsBin, "unused mocked executable"); + const originalPlatform = Object.getOwnPropertyDescriptor(process, "platform"); + Object.defineProperty(process, "platform", { value: "win32" }); + try { + await ctx.scan(); + assert.equal(ctx.calls[0].command, windowsBin); + } finally { + Object.defineProperty(process, "platform", originalPlatform); + } +}); + +test("uses an absolute operator override and preserves URL targets", async (t) => { + const ctx = await setup(t); + process.env.SKILLSPECTOR_BIN = join(ctx.root, "custom-cli"); + writeFileSync(process.env.SKILLSPECTOR_BIN, "unused"); + await ctx.scan({ target: "https://example.test/skill", noLlm: false, format: "json" }); + assert.equal(ctx.calls[0].command, process.env.SKILLSPECTOR_BIN); + assert.deepEqual(ctx.calls[0].args, ["scan", "https://example.test/skill", "--format", "json"]); +}); + +test("never resolves a workspace executable through PATH or a relative override", async (t) => { + const ctx = await setup(t); + rmSync(ctx.bin); + writeFileSync(join(ctx.workspace, "skillspector"), "unused attacker executable"); + for (const value of [undefined, "skillspector", "./skillspector", "../skillspector"]) { + if (value === undefined) delete process.env.SKILLSPECTOR_BIN; + else process.env.SKILLSPECTOR_BIN = value; + await assert.rejects(ctx.scan(), /absolute executable path/); + } + assert.equal(ctx.calls.length, 0); +}); + +test("publishes relative and absolute in-workspace reports and replaces regular files", async (t) => { + const ctx = await setup(t); + mkdirSync(join(ctx.workspace, "reports")); + for (const output of ["reports/result.json", join(ctx.workspace, "reports/result.json")]) { + const destination = resolve(ctx.workspace, output); + writeFileSync(destination, "old report"); + const result = await ctx.scan({ output }); + assert.equal(readFileSync(destination, "utf8"), "new report"); + assert.equal(result.details.outputPath, destination); + assert.notEqual(ctx.calls.at(-1).output, destination); + assert.equal(existsSync(dirname(ctx.calls.at(-1).output)), false); + assert.deepEqual(readdirSync(dirname(destination)), ["result.json"]); + } +}); + +test("new and replaced reports remain private under a permissive umask", { skip: process.platform === "win32" }, async (t) => { + const previousUmask = process.umask(0o022); + t.after(() => process.umask(previousUmask)); + const ctx = await setup(t); + writeFileSync(join(ctx.workspace, "existing.txt"), "private report", { mode: 0o600 }); + for (const output of ["existing.txt", "new.txt"]) { + await ctx.scan({ output }); + assert.equal(statSync(join(ctx.workspace, output)).mode & 0o777, 0o600); + assert.equal(readFileSync(join(ctx.workspace, output), "utf8"), "new report"); + } +}); + +test("rejects absolute and parent-relative escapes before invoking CLI", async (t) => { + const ctx = await setup(t); + const outside = join(ctx.root, "outside.txt"); + writeFileSync(outside, "preserve"); + for (const output of [outside, "../outside.txt", "../workspace-other/report", "."]) { + await assert.rejects(ctx.scan({ output }), /within the current workspace/); + } + assert.equal(readFileSync(outside, "utf8"), "preserve"); + assert.equal(ctx.calls.length, 0); +}); + +test("rejects symlinked parents, symlink files, and dangling symlinks", async (t) => { + const ctx = await setup(t); + const outside = join(ctx.root, "outside.txt"); + writeFileSync(outside, "preserve"); + symlinkSync(ctx.root, join(ctx.workspace, "escape")); + symlinkSync(outside, join(ctx.workspace, "linked.txt")); + symlinkSync(join(ctx.root, "missing.txt"), join(ctx.workspace, "dangling.txt")); + for (const output of ["escape/outside.txt", "linked.txt", "dangling.txt"]) { + await assert.rejects(ctx.scan({ output }), /within the current workspace|regular file/); + } + assert.equal(readFileSync(outside, "utf8"), "preserve"); + assert.equal(existsSync(join(ctx.root, "missing.txt")), false); + assert.equal(ctx.calls.length, 0); +}); + +test("replacing a hardlinked report does not alter the outside inode", async (t) => { + const ctx = await setup(t); + const outside = join(ctx.root, "outside.txt"); + writeFileSync(outside, "preserve"); + linkSync(outside, join(ctx.workspace, "report.txt")); + await ctx.scan({ output: "report.txt" }); + assert.equal(readFileSync(outside, "utf8"), "preserve"); + assert.equal(readFileSync(join(ctx.workspace, "report.txt"), "utf8"), "new report"); +}); + +test("rechecks a report parent changed during the scan and cleans private output", async (t) => { + const ctx = await setup(t, ({ root, workspace }) => { + renameSync(join(workspace, "reports"), join(workspace, "original-reports")); + symlinkSync(root, join(workspace, "reports")); + }); + mkdirSync(join(ctx.workspace, "reports")); + await assert.rejects(ctx.scan({ output: "reports/result.txt" }), /within the current workspace/); + assert.equal(existsSync(join(ctx.root, "result.txt")), false); + assert.equal(existsSync(dirname(ctx.calls[0].output)), false); +}); + +test("rejects a report replaced by a symlink during the scan", async (t) => { + const ctx = await setup(t, ({ root, workspace }) => { + rmSync(join(workspace, "report.txt"), { force: true }); + symlinkSync(join(root, "outside.txt"), join(workspace, "report.txt")); + }); + writeFileSync(join(ctx.root, "outside.txt"), "preserve"); + await assert.rejects(ctx.scan({ output: "report.txt" }), /regular file/); + assert.equal(readFileSync(join(ctx.root, "outside.txt"), "utf8"), "preserve"); + assert.equal(existsSync(dirname(ctx.calls[0].output)), false); +}); + +test("preserves generated reports when the scanner returns exit 1 or 2", async (t) => { + for (const code of [1, 2]) { + await t.test(`exit ${code}`, async (t) => { + const report = JSON.stringify({ execution_successful: code !== 2 }); + const ctx = await setup(t, ({ output }) => { + writeFileSync(output, report); + return { code, stderr: "scan failed" }; + }); + writeFileSync(join(ctx.workspace, "existing.json"), "old report"); + for (const output of ["existing.json", "new.json"]) { + await assert.rejects(ctx.scan({ format: "json", output }), new RegExp(`exit code ${code}`)); + const destination = join(ctx.workspace, output); + assert.equal(readFileSync(destination, "utf8"), report); + if (process.platform !== "win32") assert.equal(statSync(destination).mode & 0o777, 0o600); + assert.equal(existsSync(dirname(ctx.calls.at(-1).output)), false); + } + assert.deepEqual(readdirSync(ctx.workspace), ["existing.json", "new.json"]); + }); + } +}); + +test("cleans staged reports after operational failure, cancellation, and missing output", async (t) => { + for (const failure of ["status", "empty", "symlink", "cancel", "missing", "killed"]) { + await t.test(failure, async (t) => { + const ctx = await setup(t, ({ output, workspace }) => { + if (failure === "cancel") throw new Error("cancelled"); + if (["status", "missing", "symlink"].includes(failure)) rmSync(output); + if (failure === "empty") writeFileSync(output, ""); + if (failure === "symlink") symlinkSync(join(workspace, "report.txt"), output); + return { code: failure === "missing" ? 0 : failure === "killed" ? 137 : 2, stderr: "synthetic failure" }; + }); + writeFileSync(join(ctx.workspace, "report.txt"), "preserve"); + await assert.rejects(ctx.scan({ output: "report.txt" })); + assert.equal(readFileSync(join(ctx.workspace, "report.txt"), "utf8"), "preserve"); + assert.equal(existsSync(dirname(ctx.calls[0].output)), false); + assert.deepEqual(readdirSync(ctx.workspace), ["report.txt"]); + }); + } +}); diff --git a/tests/unit/test_suppression.py b/tests/unit/test_suppression.py index d2a6c6d8a..fede5eb22 100644 --- a/tests/unit/test_suppression.py +++ b/tests/unit/test_suppression.py @@ -17,12 +17,14 @@ from __future__ import annotations +import os from dataclasses import replace from pathlib import Path import pytest import yaml +from skillspector import suppression as suppression_module from skillspector.models import Finding from skillspector.suppression import ( SHIPPED_BASELINE_FILENAME, @@ -439,6 +441,21 @@ def test_baseline_from_dict_rejects_non_mapping() -> None: baseline_from_dict(["not", "a", "mapping"]) # type: ignore[arg-type] +@pytest.mark.parametrize("field", ["rules", "fingerprints"]) +@pytest.mark.parametrize("value", [{}, "", 0, False]) +def test_baseline_from_dict_rejects_falsy_non_list_collections(field: str, value: object) -> None: + with pytest.raises(ValueError, match="rules and fingerprints must be lists"): + baseline_from_dict({"version": 2, field: value}) + + +@pytest.mark.parametrize( + "collections", + [{}, {"rules": None}, {"fingerprints": None}, {"rules": []}, {"fingerprints": []}], +) +def test_baseline_from_dict_accepts_empty_collections(collections: dict[str, object]) -> None: + assert baseline_from_dict({"version": 2, **collections}).is_empty() + + def test_baseline_from_dict_rejects_legacy_v1_fingerprints() -> None: with pytest.raises(ValueError, match="Version 1 fingerprints cannot be trusted"): baseline_from_dict( @@ -519,6 +536,167 @@ def test_load_baseline_missing_file(tmp_path: Path) -> None: load_baseline(tmp_path / "nope.yaml") +def test_load_baseline_byte_limit_before_parsing( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(suppression_module, "MAX_BASELINE_BYTES", 32, raising=False) + out = tmp_path / "baseline.yaml" + out.write_bytes(b"version: 2\n".ljust(32, b" ")) + assert load_baseline(out).is_empty() + out.write_bytes(out.read_bytes() + b" ") + + def unexpected_parse(*args: object, **kwargs: object) -> None: + pytest.fail("oversized baseline reached YAML parsing") + + monkeypatch.setattr(yaml, "load", unexpected_parse) + with pytest.raises(ValueError, match="byte limit"): + load_baseline(out) + + +@pytest.mark.parametrize( + ("limit_name", "boundary", "message"), + [ + ("MAX_BASELINE_NODES", 3, "node limit"), + ("MAX_BASELINE_DEPTH", 2, "depth limit"), + ("MAX_BASELINE_SCALAR_CHARS", 7, "scalar character limit"), + ], +) +def test_load_baseline_yaml_limits( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + limit_name: str, + boundary: int, + message: str, +) -> None: + out = tmp_path / "baseline.yaml" + out.write_text("version: 2\n", encoding="utf-8") + monkeypatch.setattr(suppression_module, limit_name, boundary, raising=False) + assert load_baseline(out).is_empty() + monkeypatch.setattr(suppression_module, limit_name, boundary - 1, raising=False) + with pytest.raises(ValueError, match=message): + load_baseline(out) + + +def test_load_baseline_counts_alias_expansion( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + out = tmp_path / "baseline.yaml" + out.write_text( + "version: 2\nrules:\n - &rule {id: SQP-1, reason: accepted}\n - *rule\n", + encoding="utf-8", + ) + monkeypatch.setattr(suppression_module, "MAX_BASELINE_NODES", 15, raising=False) + assert len(load_baseline(out).rules) == 2 + monkeypatch.setattr(suppression_module, "MAX_BASELINE_NODES", 14, raising=False) + with pytest.raises(ValueError, match="expanded YAML node limit"): + load_baseline(out) + + +def test_load_baseline_bounds_merge_expansion_before_construction( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + out = tmp_path / "baseline.yaml" + out.write_text( + "version: 2\n" + "a: &a {id: SQP-1, reason: accepted}\n" + "b: &b {<<: [*a, *a]}\n" + "c: &c {<<: [*b, *b]}\n" + "rules: [{<<: [*c, *c]}]\n", + encoding="utf-8", + ) + monkeypatch.setattr(suppression_module, "MAX_BASELINE_NODES", 100, raising=False) + + def unexpected_construction(*args: object, **kwargs: object) -> None: + pytest.fail("expanded baseline reached object construction") + + monkeypatch.setattr(yaml.SafeLoader, "construct_document", unexpected_construction) + with pytest.raises(ValueError, match="expanded YAML node limit"): + load_baseline(out) + + +@pytest.mark.parametrize( + ("payload", "message"), + [ + ("note: &cycle [*cycle]\n", "cyclic YAML aliases"), + ( + "a: &a [value]\nb: &b [*a]\nc: [*b]\n", + "expanded YAML depth limit", + ), + ], +) +def test_load_baseline_rejects_cyclic_or_deep_aliases( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, payload: str, message: str +) -> None: + monkeypatch.setattr(suppression_module, "MAX_BASELINE_DEPTH", 4, raising=False) + out = tmp_path / "baseline.yaml" + out.write_text("version: 2\n" + payload, encoding="utf-8") + with pytest.raises(ValueError, match=message): + load_baseline(out) + + +def test_load_baseline_bounds_repeated_scalar_content( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(suppression_module, "MAX_BASELINE_BYTES", 96, raising=False) + out = tmp_path / "baseline.yaml" + content = f"version: 2\na: &text {'x' * 40}\nb: *text\nc: *text\n" + assert len(content.encode()) <= 96 + out.write_text(content, encoding="utf-8") + with pytest.raises(ValueError, match="expanded YAML character limit"): + load_baseline(out) + + +def test_load_baseline_preserves_bounded_yaml_merges(tmp_path: Path) -> None: + out = tmp_path / "baseline.yaml" + out.write_text( + "version: 2\n" + "defaults: &defaults {id: SQP-1, reason: accepted}\n" + "rules:\n - <<: *defaults\n path: SKILL.md\n" + " - <<: [*defaults, {id: SSD-2, reason: second}]\n", + encoding="utf-8", + ) + baseline = load_baseline(out) + assert baseline.rules == [ + SuppressionRule(rule_id="SQP-1", reason="accepted", path="SKILL.md"), + SuppressionRule(rule_id="SQP-1", reason="accepted"), + ] + + +def test_baseline_combined_record_limit(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(suppression_module, "MAX_BASELINE_RECORDS", 2, raising=False) + data = { + "version": 2, + "scanner_version": SCANNER_VERSION, + "rules": [{"id": "SQP-1", "reason": "accepted"}], + "fingerprints": [{"hash": "sha256:" + "a" * 64, "reason": "accepted"}], + } + baseline = baseline_from_dict(data) + assert len(baseline.rules) + len(baseline.fingerprints) == 2 + data["rules"].append({"id": "SSD-2", "reason": "accepted"}) + with pytest.raises(ValueError, match="record limit"): + baseline_from_dict(data) + out = tmp_path / "baseline.yaml" + out.write_text(yaml.safe_dump(data), encoding="utf-8") + with pytest.raises(ValueError, match="record limit"): + load_baseline(out) + + +@pytest.mark.skipif(not hasattr(os, "mkfifo"), reason="requires POSIX FIFO support") +def test_load_baseline_rejects_fifo_without_reading( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + out = tmp_path / "baseline.yaml" + os.mkfifo(out) + + def unexpected_legacy_read(*args: object, **kwargs: object) -> None: + pytest.fail("non-regular baseline reached an unbounded read") + + # Keep this regression safe on revisions that still use the blocking reader. + monkeypatch.setattr(Path, "read_text", unexpected_legacy_read) + with pytest.raises(ValueError, match="regular file"): + load_baseline(out) + + def test_build_dump_load_round_trip(tmp_path: Path) -> None: findings = [_finding(), _finding(rule_id="SDI-2", file="x/SKILL.md")] file_cache = {