Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions docs/ANALYSIS_RESOURCE_BOUNDS.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.

Expand Down
53 changes: 51 additions & 2 deletions src/skillspector/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
from __future__ import annotations

import json
import math
import os
import sys
from copy import deepcopy
Expand Down Expand Up @@ -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")
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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."""
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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")
Expand Down Expand Up @@ -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)

Expand Down
177 changes: 177 additions & 0 deletions tests/unit/test_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down
Loading