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
29 changes: 29 additions & 0 deletions contrib/batch_scan/runner.py
Original file line number Diff line number Diff line change
Expand Up @@ -779,6 +779,35 @@ def run_one(
entry["issues"] = list(entry.get("issues", [])) + annotate_findings(
[finding.to_dict() for finding in gap_findings], detected_language
)
if gap_findings:
from skillspector.nodes.report import _compute_risk_score

findings = result.get(
"active_findings",
result.get("filtered_findings", result.get("findings", [])),
)
if not isinstance(findings, list):
findings = []
component_metadata = result.get("component_metadata")
if not isinstance(component_metadata, list):
component_metadata = []
score, severity, recommendation = _compute_risk_score(
findings + gap_findings,
bool(result.get("has_executable_scripts", False)),
component_metadata,
)
risk_assessment = entry.get("risk_assessment")
if (
recommendation == "SAFE"
Comment thread
efegokdemir marked this conversation as resolved.
and isinstance(risk_assessment, dict)
and risk_assessment.get("recommendation") == "CAUTION"
):
recommendation = "CAUTION"
entry["risk_assessment"] = {
"score": score,
"severity": severity,
"recommendation": recommendation.replace("_", " "),
}
entry["enhancements"]["gap_fill_applied"] = True
entry["enhancements"]["gap_fill_findings"] = len(gap_findings)
return entry, None
Expand Down
148 changes: 148 additions & 0 deletions tests/test_batch_scan_security.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
from contrib.batch_scan import batch_scan, runner
from contrib.batch_scan.gap_fill import run_gap_fill
from skillspector import llm_analyzer_base
from skillspector.models import Finding
from skillspector.nodes.build_context import build_context

_SECRET = "DUMMY_EXTERNAL_SECRET_NEVER_SEND"
Expand Down Expand Up @@ -128,6 +129,153 @@ def invoke(prompt):
assert entry["issues"] == []


def test_gap_fill_findings_update_risk_report_and_exit_status(
batch_skill, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
) -> None:
skill, _ = batch_skill
original_result = {
"active_findings": [],
"filtered_findings": [],
"risk_score": 0,
"risk_severity": "LOW",
"risk_recommendation": "SAFE",
"component_metadata": [],
"has_executable_scripts": False,
}
_mock_scan(monkeypatch, lambda context: context.update(original_result))
findings = [
Finding(
rule_id=rule_id,
message=f"{rule_id} gap-fill finding",
severity="HIGH",
confidence=1.0,
file="SKILL.md",
category="Security",
)
for rule_id in ("P5", "P6", "P7")
]
monkeypatch.setattr(runner, "run_gap_fill", lambda *args, **kwargs: findings)
entry, error = runner.run_one(
skill,
skill.parent,
use_llm=True,
detected_language="zh",
apply_gap_fill=True,
)

assert error is None
assert entry["risk_assessment"] == {
"score": 75,
"severity": "HIGH",
"recommendation": "DO NOT INSTALL",
}
assert entry["enhancements"]["gap_fill_findings"] == 3

monkeypatch.setattr(batch_scan, "discover_skills", lambda root: [skill])
monkeypatch.setattr(
batch_scan,
"_scan_skill_bounded",
lambda *args, **kwargs: (entry, None, skill.name),
)
monkeypatch.setattr(batch_scan, "create_api_key_pool_from_env", lambda: None)
monkeypatch.setattr(
sys,
"argv",
["batch_scan", str(skill.parent), "--workers", "1", "-f", "markdown"],
)

with pytest.raises(SystemExit) as exited:
batch_scan._main_impl()

assert exited.value.code == 1
output = capsys.readouterr().out
assert "75/100 HIGH" in output
assert "HIGH / CRITICAL Issue Details" in output
assert "P5" in output and "P6" in output and "P7" in output


def test_empty_gap_fill_preserves_core_risk_assessment(
batch_skill, monkeypatch: pytest.MonkeyPatch
) -> None:
skill, _ = batch_skill
findings = [
Finding(
rule_id=rule_id,
message=f"{rule_id} core finding",
severity="MEDIUM",
confidence=1.0,
file="SKILL.md",
category="Security",
)
for rule_id in ("P5", "P6", "P7")
]
original_result = {
"active_findings": findings,
"filtered_findings": findings,
"risk_score": 30,
"risk_severity": "MEDIUM",
"risk_recommendation": "CAUTION",
}
_mock_scan(monkeypatch, lambda context: context.update(original_result))
monkeypatch.setattr(runner, "run_gap_fill", lambda *args, **kwargs: [])

entry, error = runner.run_one(
skill,
skill.parent,
use_llm=True,
detected_language="zh",
apply_gap_fill=True,
)

assert error is None
assert entry["risk_assessment"] == {
"score": 30,
"severity": "MEDIUM",
"recommendation": "CAUTION",
}
assert entry["enhancements"]["gap_fill_applied"] is True
assert entry["enhancements"]["gap_fill_findings"] == 0


def test_gap_fill_preserves_caution_for_incomplete_scan(
batch_skill, monkeypatch: pytest.MonkeyPatch
) -> None:
skill, _ = batch_skill
original_result = {
"active_findings": [],
"filtered_findings": [],
"risk_score": 0,
"risk_severity": "LOW",
"risk_recommendation": "CAUTION",
"component_metadata": [],
"has_executable_scripts": False,
}
_mock_scan(monkeypatch, lambda context: context.update(original_result))
finding = Finding(
rule_id="P5",
message="Low-severity gap-fill finding",
severity="LOW",
confidence=1.0,
file="SKILL.md",
category="Security",
)
monkeypatch.setattr(runner, "run_gap_fill", lambda *args, **kwargs: [finding])

entry, error = runner.run_one(
skill,
skill.parent,
use_llm=True,
detected_language="zh",
apply_gap_fill=True,
)

assert error is None
assert entry["risk_assessment"]["score"] > 0
assert entry["risk_assessment"]["severity"] == "LOW"
assert entry["risk_assessment"]["recommendation"] == "CAUTION"
assert entry["enhancements"]["gap_fill_findings"] == 1


@pytest.mark.parametrize("cache_state", ["empty", "missing"])
def test_gap_fill_never_falls_back_to_local_content(
batch_skill, monkeypatch: pytest.MonkeyPatch, cache_state
Expand Down
Loading