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
9 changes: 4 additions & 5 deletions src/skillspector/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,7 @@
build_baseline_dict,
discover_baseline,
dump_baseline,
effective_findings,
load_baseline,
)

Expand Down Expand Up @@ -510,8 +511,7 @@ def _scan_multi_skill(
continue
score = result.get("risk_score", 0)
severity = result.get("risk_severity", "LOW")
filtered = result.get("filtered_findings") or result.get("findings")
finding_count = len(filtered) if isinstance(filtered, list) else 0
finding_count = len(effective_findings(result))
execution = "failed" if result.get("execution_successful") is False else "successful"
console.print(
f" {skill.name:<30} {score:<8} {severity:<12} {finding_count:<10} {execution:<10}"
Expand All @@ -533,8 +533,7 @@ def _scan_multi_skill(
combined_skills.append({"name": skill.name, "error": result["error"]})
else:
payload = _recursive_json_payload(result) or {}
selected_findings = result.get("filtered_findings") or result.get("findings") or []
finding_count = len(selected_findings) if isinstance(selected_findings, list) else 0
finding_count = len(effective_findings(result))
entry = {
"name": skill.name,
"path": skill.relative_path,
Expand Down Expand Up @@ -668,7 +667,7 @@ def baseline(
state = _scan_state(input_path, FormatChoice.json, no_llm)
state["baseline_path"] = os.path.abspath(output.expanduser())
result = graph.invoke(state)
findings = result.get("filtered_findings") or result.get("findings") or []
findings = effective_findings(result)
data = build_baseline_dict(
findings,
reason=reason,
Expand Down
3 changes: 2 additions & 1 deletion src/skillspector/mcp_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@
from skillspector.graph import graph
from skillspector.llm_utils import is_llm_available
from skillspector.logging_config import get_logger
from skillspector.suppression import effective_findings

if TYPE_CHECKING:
from mcp.server.fastmcp import FastMCP
Expand Down Expand Up @@ -137,7 +138,7 @@ async def run_scan(
},
},
)
findings = result.get("filtered_findings") or result.get("findings") or []
findings = effective_findings(result)
risk_score = int(result.get("risk_score") or 0)
execution_successful = bool(result.get("execution_successful", True))
analysis_completeness = result.get("analysis_completeness") or {}
Expand Down
43 changes: 43 additions & 0 deletions src/skillspector/suppression.py
Original file line number Diff line number Diff line change
Expand Up @@ -387,6 +387,49 @@ def partition_findings(
return kept, suppressed


def effective_findings(result: Mapping[str, object]) -> list[Finding]:
"""Return the findings from a graph *result* that actually drove its risk score.

The report node returns ``filtered_findings`` as the full pre-partition set
(kept plus baseline-suppressed) alongside ``suppressed_findings``, but scores
and SARIF results from the kept subset alone. Consumers that want the numbers
the report itself published must therefore subtract the suppressed partition.

Two failure modes this exists to prevent, both of which over-report:

* ``result.get("filtered_findings") or result.get("findings")`` treats an
empty filtered list as absent and falls back to the raw pre-filter
findings. An empty list is a real answer -- every finding was filtered out
or suppressed -- not a missing one.
* Using ``filtered_findings`` directly counts baseline-suppressed findings
that the report excluded from the score, so a fully suppressed skill
reports risk 0 alongside a non-zero finding count.

Falls back to the raw ``findings`` list only when ``filtered_findings`` is
absent or malformed, and does not subtract there: raw findings are not the
population that produced ``suppressed_findings``.
"""
filtered = result.get("filtered_findings")
if not isinstance(filtered, list):
raw = result.get("findings")
return list(raw) if isinstance(raw, list) else []

suppressed = result.get("suppressed_findings")
if not isinstance(suppressed, list) or not suppressed:
return list(filtered)

suppressed_ids = {
entry.finding.finding_id
for entry in suppressed
if isinstance(entry, SuppressedFinding) and entry.finding is not None
}
return [
finding
for finding in filtered
if not isinstance(finding, Finding) or finding.finding_id not in suppressed_ids
]


def build_baseline_dict(
findings: list[Finding],
reason: str = "Accepted finding (auto-generated baseline)",
Expand Down
139 changes: 139 additions & 0 deletions tests/unit/test_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
"""Tests for skillspector CLI (skillspector scan, --version)."""

import json
import re
from pathlib import Path
from types import SimpleNamespace
from typing import Any
Expand All @@ -28,7 +29,9 @@

from skillspector import __version__
from skillspector.cli import FormatChoice, _scan_multi_skill, app
from skillspector.models import Finding
from skillspector.multi_skill import MultiSkillDetectionResult, SkillDirectory
from skillspector.suppression import SuppressedFinding

runner = CliRunner()

Expand Down Expand Up @@ -1133,3 +1136,139 @@ def fake_invoke(state: dict[str, Any], config: Any = None) -> dict[str, Any]:
assert payload["issues"] == [{"id": "X-1", "severity": "low"}]
assert payload["suppressed_count"] == 0
assert payload["suppressed"] == []


def _combined_json_counts(results: list[dict[str, Any]], tmp_path: Path) -> list[int]:
"""Run a recursive JSON scan over stubbed results and return per-skill counts."""
skills = [
SkillDirectory(path=tmp_path / f"skill{i}", name=f"skill{i}", relative_path=f"skill{i}")
for i in range(1, len(results) + 1)
]
detection = MultiSkillDetectionResult(is_multi_skill=True, skills=skills, has_root_skill=False)
out = tmp_path / "combined.json"

with patch("skillspector.cli.graph.invoke", side_effect=results):
_scan_multi_skill(
detection, FormatChoice.json, out, no_llm=True, yara_rules_dir=None, verbose=False
)

data = json.loads(out.read_text(encoding="utf-8"))
return [entry["finding_count"] for entry in data["skills"]]


def test_cli_recursive_json_count_excludes_suppressed_findings(tmp_path: Path) -> None:
"""Combined JSON counts the active findings, not the pre-partition set.

`report` returns `filtered_findings` as kept+suppressed and scores only the
kept subset, so counting `filtered_findings` made a fully suppressed
sub-skill report risk 0 alongside a non-zero finding count.
"""
findings = [
Finding(rule_id="SQP-1", message="one"),
Finding(rule_id="SQP-2", message="two"),
Finding(rule_id="SQP-3", message="three"),
]
fully_suppressed = {
"report_body": "{}",
"risk_score": 0,
"risk_severity": "LOW",
"findings": list(findings),
"filtered_findings": list(findings),
"suppressed_findings": [
SuppressedFinding(finding=finding, reason="baselined") for finding in findings
],
}
partly_suppressed = {
"report_body": "{}",
"risk_score": 20,
"risk_severity": "LOW",
"findings": list(findings),
"filtered_findings": list(findings),
"suppressed_findings": [
SuppressedFinding(finding=finding, reason="baselined") for finding in findings[:2]
],
}

assert _combined_json_counts([fully_suppressed, partly_suppressed], tmp_path) == [0, 1]


def test_cli_recursive_json_count_respects_an_empty_filtered_list(tmp_path: Path) -> None:
"""Every-finding-filtered is reported as 0, not as the raw pre-filter count."""
result = {
"report_body": "{}",
"risk_score": 0,
"risk_severity": "LOW",
"findings": [Finding(rule_id="SQP-1", message="one")],
"filtered_findings": [],
"suppressed_findings": [],
}

assert _combined_json_counts([result], tmp_path) == [0]


def test_cli_recursive_summary_count_excludes_suppressed(
tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
"""The terminal summary's Findings column uses the same active count.

Pinned separately from the JSON path: the two call sites are independent
lines, so a regression in one is invisible to a test covering the other.
"""
findings = [Finding(rule_id="SQP-1", message="one"), Finding(rule_id="SQP-2", message="two")]
result = {
"report_body": "# report",
"risk_score": 0,
"risk_severity": "LOW",
"findings": list(findings),
"filtered_findings": list(findings),
"suppressed_findings": [
SuppressedFinding(finding=finding, reason="baselined") for finding in findings
],
}
detection = MultiSkillDetectionResult(
is_multi_skill=True,
skills=[SkillDirectory(path=tmp_path / "solo", name="solo", relative_path="solo")],
has_root_skill=False,
)

with patch("skillspector.cli.graph.invoke", side_effect=[result]):
_scan_multi_skill(
detection, FormatChoice.terminal, None, no_llm=True, yara_rules_dir=None, verbose=False
)

summary = re.sub(r"\x1b\[[0-9;]*m", "", capsys.readouterr().out)
row = next(line for line in summary.splitlines() if line.strip().startswith("solo"))
assert row.split() == ["solo", "0", "LOW", "0", "successful"]


def test_cli_baseline_command_excludes_filtered_out_findings(tmp_path: Path) -> None:
"""`skillspector baseline` fingerprints what the scan reported, not raw findings.

Closes a mutation survivor: reverting this call site to the old
`filtered_findings or findings` passed the entire suite, because nothing
drove the baseline command through an empty filtered list. An empty filtered
list means every finding was filtered out, so building a baseline from the
raw list would write fingerprints suppressing findings the scan never
reported, and would fail closed on the next run for no reason.
"""
skill = tmp_path / "skill"
skill.mkdir()
source = "---\nname: b\n---\nbody\n"
(skill / "SKILL.md").write_text(source, encoding="utf-8")
out = tmp_path / "baseline.yaml"

result = {
"findings": [Finding(rule_id="SQP-1", message="one", file="SKILL.md")],
"filtered_findings": [],
"suppressed_findings": [],
"file_cache": {"SKILL.md": source},
"risk_score": 0,
}

with patch("skillspector.cli.graph.invoke", return_value=result):
invocation = runner.invoke(app, ["baseline", str(skill), "-o", str(out), "--no-llm"])

assert invocation.exit_code == 0, invocation.output
written = yaml.safe_load(out.read_text(encoding="utf-8"))
assert written.get("fingerprints", []) == []
assert "0 suppressed finding(s)" in re.sub(r"\x1b\[[0-9;]*m", "", invocation.output)
47 changes: 47 additions & 0 deletions tests/unit/test_mcp_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,9 @@

from skillspector import mcp_server
from skillspector.mcp_server import run_scan
from skillspector.models import Finding
from skillspector.providers import reset_provider, use_provider
from skillspector.suppression import SuppressedFinding


def _write_skill(tmp_path: Path, body: str = "# Safe skill") -> Path:
Expand Down Expand Up @@ -478,3 +480,48 @@ async def test_mcp_stdio_initialize_registers_scan_skill() -> None:
tools = await asyncio.wait_for(session.list_tools(), timeout=15)

assert "scan_skill" in {tool.name for tool in tools.tools}


async def test_run_scan_findings_exclude_the_suppressed_partition(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""The MCP verdict lists the findings that drove the score, not kept+suppressed.

`run_scan` serialises this list straight to the calling agent, so a
baseline-suppressed finding leaking in tells the agent a skill is dirtier
than the risk score it is gating on.
"""
kept = Finding(rule_id="SQP-1", message="kept")
dropped = Finding(rule_id="SQP-2", message="suppressed")
result = {
"findings": [kept, dropped],
"filtered_findings": [kept, dropped],
"suppressed_findings": [SuppressedFinding(finding=dropped, reason="baselined")],
"risk_score": 10,
"risk_severity": "LOW",
"report_body": "# report",
}
monkeypatch.setattr(mcp_server.graph, "ainvoke", AsyncMock(return_value=result))

verdict = await run_scan(str(_write_skill(tmp_path)), use_llm=False, output_format="json")

assert [finding["id"] for finding in verdict["findings"]] == ["SQP-1"]


async def test_run_scan_respects_an_empty_filtered_list(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Every-finding-filtered reports no findings, not the raw pre-filter list."""
result = {
"findings": [Finding(rule_id="SQP-1", message="one")],
"filtered_findings": [],
"suppressed_findings": [],
"risk_score": 0,
"risk_severity": "LOW",
"report_body": "# report",
}
monkeypatch.setattr(mcp_server.graph, "ainvoke", AsyncMock(return_value=result))

verdict = await run_scan(str(_write_skill(tmp_path)), use_llm=False, output_format="json")

assert verdict["findings"] == []
Loading
Loading