diff --git a/docs/ANALYSIS_RESOURCE_BOUNDS.md b/docs/ANALYSIS_RESOURCE_BOUNDS.md index ea556630..9366eb53 100644 --- a/docs/ANALYSIS_RESOURCE_BOUNDS.md +++ b/docs/ANALYSIS_RESOURCE_BOUNDS.md @@ -206,6 +206,8 @@ When relevant analysis is incomplete: - `skillspector scan --fail-on-incomplete` exits with status 1. Without this option, the CLI retains its compatibility behavior and still applies its ordinary risk-score exit policy. Execution failures exit with status 2. +- `skillspector scan --min-coverage PERCENT` exits with status 1 when canonical coverage is below + `PERCENT`; equality passes and recursive scans evaluate each child. - MCP responses set `safe_to_install` to `false` when analysis is incomplete, any relevant file is entirely uninspected, execution failed, or the risk score exceeds the installation threshold. diff --git a/src/skillspector/cli.py b/src/skillspector/cli.py index 4bb3e34d..aff9a71a 100644 --- a/src/skillspector/cli.py +++ b/src/skillspector/cli.py @@ -22,6 +22,7 @@ from __future__ import annotations import json +import math import os import sys from copy import deepcopy @@ -332,6 +333,28 @@ def _write_result( print(report_body) +def _validate_min_coverage(value: float | None) -> float | None: + """Accept only finite percentage thresholds in the CLI's supported range.""" + if value is not None and (not math.isfinite(value) or not 0 <= value <= 100): + raise typer.BadParameter("must be a finite number between 0 and 100") + return value + + +def _coverage_below_threshold(result: dict[str, object], threshold: float | None) -> bool: + """Fail closed when an enabled threshold has no numeric canonical coverage.""" + if threshold is None: + return False + completeness = result.get("analysis_completeness") + if not isinstance(completeness, dict): + return True + coverage = completeness.get("coverage_percent") + if isinstance(coverage, bool) or not isinstance(coverage, (int, float)): + return True + if not math.isfinite(float(coverage)): + return True + return float(coverage) < threshold + + def _recursive_json_payload(result: dict[str, object]) -> dict[str, object] | None: """Return parsed report_body when it is valid JSON object text.""" raw_report_body = result.get("report_body") @@ -494,6 +517,14 @@ def scan( help="Exit 1 when relevant analysis is partial or incomplete.", ), ] = False, + min_coverage: Annotated[ + float | None, + typer.Option( + "--min-coverage", + help="Exit 1 when canonical analysis coverage is below this percentage (0-100).", + callback=_validate_min_coverage, + ), + ] = None, verbose: Annotated[ bool, typer.Option( @@ -541,10 +572,17 @@ def scan( 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: + if ( + recursive + or baseline is not None + or show_suppressed + or yara_rules_dir is not None + or min_coverage is not None + ): err_console.print( "[red]Error:[/red] --mcp-registry cannot be combined with " - "--recursive, --baseline, --show-suppressed, or --yara-rules-dir" + "--recursive, --baseline, --show-suppressed, --yara-rules-dir, " + "or --min-coverage" ) raise typer.Exit(code=2) if format != FormatChoice.json: @@ -617,6 +655,7 @@ def scan( yara_dir=yara_dir, verbose=verbose, fail_on_incomplete=fail_on_incomplete, + min_coverage=min_coverage, ) return if detection.complete and not detection.has_root_skill and len(detection.skills) == 0: @@ -691,6 +730,8 @@ def scan( ) if fail_on_incomplete and not is_complete: raise typer.Exit(code=1) + if _coverage_below_threshold(result, min_coverage): + raise typer.Exit(code=1) if (result.get("risk_score") or 0) > RISK_THRESHOLD: raise typer.Exit(code=1) except typer.Exit: @@ -2064,6 +2105,7 @@ def _scan_multi_skill( yara_dir: str | None = None, verbose: bool = False, fail_on_incomplete: bool = False, + min_coverage: float | None = None, **legacy_kwargs: object, ) -> None: """Scan each detected sub-skill independently and produce a combined report.""" @@ -2095,6 +2137,7 @@ def _scan_multi_skill( complete_skill_count = 0 partial_skill_count = 0 failed_skill_count = 0 + coverage_failed = False for i, skill in enumerate(skills, 1): if i > _MULTI_SKILL_MAX_SKILLS: @@ -2170,6 +2213,8 @@ def _scan_multi_skill( if child_failed: execution_failed = True failed_skill_count += 1 + elif _coverage_below_threshold(result, min_coverage): + coverage_failed = True completeness_value = result.get("analysis_completeness") if ( not child_failed @@ -2213,6 +2258,8 @@ def _scan_multi_skill( omitted_skills=omitted_skill_count, limitations=aggregate_limitations, ) + if _coverage_below_threshold({"analysis_completeness": aggregate_completeness}, min_coverage): + coverage_failed = True analysis_incomplete = not bool(aggregate_completeness["is_complete"]) console.print("\n[bold]═══ Multi-Skill Summary ═══[/bold]\n") @@ -2381,6 +2428,8 @@ def _scan_multi_skill( raise typer.Exit(code=2) if fail_on_incomplete and analysis_incomplete: raise typer.Exit(code=1) + if coverage_failed: + raise typer.Exit(code=1) if max_score > RISK_THRESHOLD: raise typer.Exit(code=1) diff --git a/tests/unit/test_cli.py b/tests/unit/test_cli.py index bbb62c6e..2109ca6f 100644 --- a/tests/unit/test_cli.py +++ b/tests/unit/test_cli.py @@ -198,6 +198,183 @@ def test_cli_fail_on_incomplete_exits_one_after_writing_report( assert output.exists() +@pytest.mark.parametrize( + ("coverage", "threshold", "exit_code"), + [(86.9, 87.0, 1), (87.0, 87.0, 0)], +) +def test_cli_min_coverage_uses_strict_boundary_and_writes_report( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + coverage: float, + threshold: float, + exit_code: int, +) -> None: + (tmp_path / "SKILL.md").write_text("# Safe", encoding="utf-8") + output = tmp_path / "report.json" + monkeypatch.setattr( + "skillspector.cli.graph.invoke", + lambda state, config: { + "report_body": json.dumps({"analysis_completeness": {"coverage_percent": coverage}}), + "execution_successful": True, + "analysis_completeness": {"coverage_percent": coverage}, + "risk_score": 0, + }, + ) + result = runner.invoke( + app, + [ + "scan", + str(tmp_path), + "-f", + "json", + "-o", + str(output), + "--min-coverage", + str(threshold), + ], + ) + assert result.exit_code == exit_code + assert json.loads(output.read_text())["analysis_completeness"]["coverage_percent"] == coverage + + +def test_cli_min_coverage_fails_closed_for_missing_coverage( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + (tmp_path / "SKILL.md").write_text("# Safe", encoding="utf-8") + output = tmp_path / "report.json" + monkeypatch.setattr( + "skillspector.cli.graph.invoke", + lambda state, config: { + "report_body": "{}", + "execution_successful": True, + "risk_score": 0, + }, + ) + result = runner.invoke( + app, + [ + "scan", + str(tmp_path), + "--no-llm", + "--min-coverage", + "87", + "--output", + str(output), + "--format", + "json", + ], + ) + assert result.exit_code == 1 + assert output.exists() + + +@pytest.mark.parametrize("value", ["-1", "101", "nan", "inf"]) +def test_cli_min_coverage_rejects_invalid_values(tmp_path: Path, value: str) -> None: + (tmp_path / "SKILL.md").write_text("# Safe", encoding="utf-8") + result = runner.invoke(app, ["scan", str(tmp_path), "--min-coverage", value]) + assert result.exit_code == 2 + plain_output = re.sub(r"\x1b\[[0-?]*[ -/]*[@-~]", "", result.output) + assert "--min-coverage" in plain_output + + +def test_cli_mcp_registry_rejects_min_coverage(tmp_path: Path) -> None: + payload = tmp_path / "registry.json" + payload.write_text('{"servers": []}', encoding="utf-8") + result = runner.invoke( + app, + ["scan", str(payload), "--mcp-registry", "--min-coverage", "87"], + ) + assert result.exit_code == 2 + assert "--min-coverage" in result.output + + +def test_recursive_min_coverage_checks_each_child_and_writes_report(tmp_path: Path) -> None: + s1 = SkillDirectory(path=tmp_path / "one", name="one", relative_path="one") + s2 = SkillDirectory(path=tmp_path / "two", name="two", relative_path="two") + detection = MultiSkillDetectionResult( + is_multi_skill=True, skills=[s1, s2], has_root_skill=False + ) + output = tmp_path / "combined.json" + with patch( + "skillspector.cli.graph.invoke", + side_effect=[ + { + "report_body": json.dumps({"analysis_completeness": {"coverage_percent": 100}}), + "analysis_completeness": {"coverage_percent": 100}, + "risk_score": 0, + }, + { + "report_body": json.dumps({"analysis_completeness": {"coverage_percent": 80}}), + "analysis_completeness": {"coverage_percent": 80}, + "risk_score": 0, + }, + ], + ): + with pytest.raises(typer.Exit) as exit_info: + _scan_multi_skill( + detection, + FormatChoice.json, + output, + no_llm=True, + min_coverage=87, + ) + assert exit_info.value.exit_code == 1 + assert output.exists() + + +def test_recursive_min_coverage_passes_when_all_children_and_aggregate_pass( + tmp_path: Path, +) -> None: + skills = [ + SkillDirectory(path=tmp_path / "one", name="one", relative_path="one"), + SkillDirectory(path=tmp_path / "two", name="two", relative_path="two"), + ] + output = tmp_path / "combined.json" + child = { + "report_body": json.dumps({"analysis_completeness": {"coverage_percent": 100}}), + "analysis_completeness": {"coverage_percent": 100}, + "risk_score": 0, + } + with patch("skillspector.cli.graph.invoke", side_effect=[child.copy(), child.copy()]): + _scan_multi_skill( + MultiSkillDetectionResult(is_multi_skill=True, skills=skills, has_root_skill=False), + FormatChoice.json, + output, + no_llm=True, + min_coverage=90, + ) + assert output.exists() + assert ( + json.loads(output.read_text(encoding="utf-8"))["analysis_completeness"]["coverage_percent"] + == 100.0 + ) + + +def test_recursive_min_coverage_fails_on_omitted_aggregate_scope( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + skills = [SkillDirectory(tmp_path / name, name, name) for name in ("one", "two", "three")] + output = tmp_path / "combined.json" + monkeypatch.setattr(cli, "_MULTI_SKILL_MAX_PUBLIC_RECORDS", 1) + monkeypatch.setattr( + cli.graph, "invoke", lambda *_args, **_kwargs: _bounded_recursive_result("one") + ) + + with pytest.raises(typer.Exit) as exit_info: + _scan_multi_skill( + MultiSkillDetectionResult(is_multi_skill=True, skills=skills), + FormatChoice.json, + output, + no_llm=True, + min_coverage=90, + ) + + assert exit_info.value.exit_code == 1 + assert json.loads(output.read_text(encoding="utf-8"))["analysis_completeness"][ + "coverage_percent" + ] == pytest.approx(33.33) + + def test_recursive_scan_exits_two_after_writing_all_child_reports(tmp_path: Path) -> None: """Recursive mode aggregates child execution failures after producing output.""" s1 = SkillDirectory(path=tmp_path / "one", name="one", relative_path="one")