b085e116c5
- Revised the metadata format in multiple SKILL.md files to use block mapping instead of single-line JSON, enhancing YAML compatibility. - Added a new entry for local agent tooling in .gitignore to exclude the .claude directory, improving project cleanliness. - Removed outdated entries for AGENTS.md and CLAUDE.md from .gitignore, streamlining ignored files.
616 lines
23 KiB
Python
616 lines
23 KiB
Python
#!/usr/bin/env python3
|
|
"""Scan all skills for security issues and produce the security scan report.
|
|
|
|
Writes a human-readable report plus a machine-readable companion. The two files
|
|
must always be generated together: the markdown links into the JSON for
|
|
per-skill scan dates, and the JSON is what the next run reads as its cache.
|
|
|
|
`SECURITY.md` is deliberately not written here: GitHub resolves that filename as
|
|
the repository's official security policy, which is hand-authored.
|
|
|
|
Two things keep the run time down. Skills are scanned concurrently, because each
|
|
scan is blocked on LLM network I/O rather than local CPU. And skills whose
|
|
content has not changed since the last published report reuse that report's
|
|
findings instead of being rescanned; a change of scanner version or model, or
|
|
`--full`, invalidates every cached result.
|
|
|
|
Environment:
|
|
SKILL_SCANNER_LLM_API_KEY API key for the LLM analyzer
|
|
SKILL_SCANNER_LLM_MODEL model id (default: claude-opus-5)
|
|
SKILL_SCAN_WORKERS concurrent skills (default: 8)
|
|
SKILL_SCAN_FULL set to 1 to force a full rescan
|
|
SKILL_SCAN_MAX_AGE_DAYS force a full rescan after this many days (default: 30)
|
|
"""
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import threading
|
|
import time
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
from datetime import datetime, timedelta, timezone
|
|
from importlib.metadata import PackageNotFoundError, version
|
|
from pathlib import Path
|
|
|
|
from dotenv import load_dotenv
|
|
from skill_scanner import SkillScanner
|
|
from skill_scanner.core.analyzers import (
|
|
BehavioralAnalyzer,
|
|
LLMAnalyzer,
|
|
TriggerAnalyzer,
|
|
)
|
|
from skill_scanner.core.loader import SkillLoadError
|
|
from skill_scanner.core.models import (
|
|
Finding,
|
|
Report,
|
|
ScanResult,
|
|
Severity,
|
|
ThreatCategory,
|
|
)
|
|
from skill_scanner.core.scan_policy import ScanPolicy
|
|
|
|
load_dotenv()
|
|
|
|
SKILLS_DIR = "skills"
|
|
OUTPUT_FILE = "docs/security-report.md"
|
|
JSON_OUTPUT_FILE = "docs/security-report.json"
|
|
|
|
SCANNER_PACKAGE = "cisco-ai-skill-scanner"
|
|
DEFAULT_WORKERS = 8
|
|
DEFAULT_MAX_AGE_DAYS = 30
|
|
|
|
|
|
def scanner_version() -> str:
|
|
try:
|
|
return version(SCANNER_PACKAGE)
|
|
except PackageNotFoundError: # pragma: no cover - editable/source installs
|
|
return "unknown"
|
|
|
|
|
|
def llm_model_id() -> str:
|
|
return os.getenv("SKILL_SCANNER_LLM_MODEL", "claude-opus-5")
|
|
|
|
|
|
def content_hash(skill_dir: Path) -> str:
|
|
"""Fingerprint everything in a skill package.
|
|
|
|
Covers file paths as well as bytes, so a rename or deletion changes the hash
|
|
even when the surviving content is identical. `__pycache__` is excluded
|
|
because it is a build artifact that varies without the skill changing.
|
|
"""
|
|
digest = hashlib.sha256()
|
|
files = sorted(
|
|
f for f in skill_dir.rglob("*")
|
|
if f.is_file() and "__pycache__" not in f.parts
|
|
)
|
|
for f in files:
|
|
digest.update(str(f.relative_to(skill_dir)).encode())
|
|
digest.update(b"\0")
|
|
digest.update(f.read_bytes())
|
|
digest.update(b"\0")
|
|
return digest.hexdigest()
|
|
|
|
|
|
def build_scanner() -> SkillScanner:
|
|
policy = ScanPolicy.from_preset("balanced")
|
|
policy.llm_analysis.max_instruction_body_chars = 75_000
|
|
policy.llm_analysis.max_referenced_file_chars = 75_000
|
|
policy.llm_analysis.max_code_file_chars = 75_000
|
|
policy.llm_analysis.max_total_prompt_chars = 500_000
|
|
llm_model = llm_model_id()
|
|
llm_key = os.getenv("SKILL_SCANNER_LLM_API_KEY")
|
|
|
|
scanner = SkillScanner(
|
|
analyzers=[
|
|
BehavioralAnalyzer(),
|
|
TriggerAnalyzer(),
|
|
LLMAnalyzer(model=llm_model, api_key=llm_key, policy=policy),
|
|
],
|
|
policy=policy,
|
|
)
|
|
return scanner
|
|
|
|
|
|
def severity_badge(sev: str) -> str:
|
|
icons = {
|
|
"CRITICAL": "🔴",
|
|
"HIGH": "🟠",
|
|
"MEDIUM": "🟡",
|
|
"LOW": "🔵",
|
|
"INFO": "⚪",
|
|
"SAFE": "🟢",
|
|
}
|
|
return f"{icons.get(sev, '⚫')} {sev}"
|
|
|
|
|
|
def generate_report(report, run_meta: dict | None = None) -> str:
|
|
now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
|
|
run_meta = run_meta or {}
|
|
lines: list[str] = []
|
|
|
|
lines.append("# Security Scan Report")
|
|
lines.append("")
|
|
lines.append(f"**Generated:** {now} ")
|
|
lines.append(f"**Skills scanned:** {report.total_skills_scanned} ")
|
|
lines.append(f"**Total findings:** {report.total_findings} ")
|
|
lines.append(f"**Critical:** {report.critical_count} | **High:** {report.high_count} | **Safe skills:** {report.safe_count}/{report.total_skills_scanned}")
|
|
lines.append("")
|
|
|
|
if run_meta:
|
|
scanner_line = (
|
|
f"**Scanner:** {SCANNER_PACKAGE} {run_meta.get('scanner_version', '?')} · "
|
|
f"**Model:** {run_meta.get('model', '?')} "
|
|
)
|
|
lines.append(scanner_line)
|
|
reused = run_meta.get("skills_reused") or 0
|
|
if reused and not run_meta.get("full_scan"):
|
|
lines.append(
|
|
f"**This run:** {run_meta.get('skills_rescanned', 0)} skill(s) rescanned; "
|
|
f"{reused} unchanged since the last scan and carried forward unmodified. "
|
|
f"Per-skill scan dates are in "
|
|
f"[`security-report.json`](security-report.json) (`last_scanned`). "
|
|
)
|
|
else:
|
|
lines.append(
|
|
f"**This run:** full rescan of all "
|
|
f"{run_meta.get('skills_rescanned', report.total_skills_scanned)} skill(s). "
|
|
)
|
|
lines.append("")
|
|
|
|
# Summary table
|
|
lines.append("## Summary")
|
|
lines.append("")
|
|
lines.append("| Skill | Severity | Findings | Safe | Duration |")
|
|
lines.append("|-------|----------|----------|------|----------|")
|
|
|
|
sorted_results = sorted(
|
|
report.scan_results,
|
|
key=lambda r: ["CRITICAL", "HIGH", "MEDIUM", "LOW", "INFO", "SAFE"].index(
|
|
r.max_severity.value if hasattr(r.max_severity, "value") else str(r.max_severity)
|
|
),
|
|
)
|
|
|
|
for result in sorted_results:
|
|
sev = result.max_severity.value if hasattr(result.max_severity, "value") else str(result.max_severity)
|
|
safe = "✅" if result.is_safe else "❌"
|
|
duration = f"{result.scan_duration_seconds:.1f}s"
|
|
lines.append(f"| {result.skill_name} | {severity_badge(sev)} | {len(result.findings)} | {safe} | {duration} |")
|
|
|
|
lines.append("")
|
|
|
|
# Per-skill details (only for skills with findings)
|
|
flagged = [r for r in sorted_results if r.findings]
|
|
if flagged:
|
|
lines.append("## Detailed Findings")
|
|
lines.append("")
|
|
|
|
for result in flagged:
|
|
sev = result.max_severity.value if hasattr(result.max_severity, "value") else str(result.max_severity)
|
|
lines.append(f"### {result.skill_name} — {severity_badge(sev)}")
|
|
lines.append("")
|
|
|
|
for finding in result.findings:
|
|
fsev = finding.severity.value if hasattr(finding.severity, "value") else str(finding.severity)
|
|
lines.append(f"- **{severity_badge(fsev)}** `{finding.rule_id}` — {finding.title}")
|
|
if finding.description:
|
|
lines.append(f" > {finding.description}")
|
|
if finding.file_path:
|
|
loc = finding.file_path
|
|
if finding.line_number:
|
|
loc += f":{finding.line_number}"
|
|
lines.append(f" > File: `{loc}`")
|
|
if finding.remediation:
|
|
lines.append(f" > **Remediation:** {finding.remediation}")
|
|
lines.append("")
|
|
|
|
else:
|
|
lines.append("## Detailed Findings")
|
|
lines.append("")
|
|
lines.append("No findings to report — all skills passed.")
|
|
lines.append("")
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
def _enum_value(value) -> str:
|
|
"""Render an enum-or-string field as a plain string."""
|
|
return value.value if hasattr(value, "value") else str(value)
|
|
|
|
|
|
def _finding_dict(finding) -> dict:
|
|
"""Serialize a finding losslessly enough to be reconstructed on a later run."""
|
|
return {
|
|
"id": finding.id,
|
|
"rule_id": finding.rule_id,
|
|
"severity": _enum_value(finding.severity),
|
|
"category": _enum_value(finding.category),
|
|
"title": finding.title,
|
|
"description": finding.description,
|
|
"file_path": finding.file_path,
|
|
"line_number": finding.line_number,
|
|
"snippet": finding.snippet,
|
|
"remediation": finding.remediation,
|
|
"analyzer": finding.analyzer,
|
|
"metadata": finding.metadata,
|
|
}
|
|
|
|
|
|
def _finding_from_dict(data: dict) -> Finding:
|
|
return Finding(
|
|
id=data.get("id", ""),
|
|
rule_id=data["rule_id"],
|
|
category=ThreatCategory(data["category"]),
|
|
severity=Severity(data["severity"]),
|
|
title=data.get("title", ""),
|
|
description=data.get("description", ""),
|
|
file_path=data.get("file_path"),
|
|
line_number=data.get("line_number"),
|
|
snippet=data.get("snippet"),
|
|
remediation=data.get("remediation"),
|
|
analyzer=data.get("analyzer"),
|
|
metadata=data.get("metadata") or {},
|
|
)
|
|
|
|
|
|
def _result_from_dict(data: dict) -> ScanResult:
|
|
"""Rebuild a ScanResult from a previous report so it can be reused as-is."""
|
|
return ScanResult(
|
|
skill_name=data["name"],
|
|
skill_directory=data["directory"],
|
|
findings=[_finding_from_dict(f) for f in data.get("findings") or []],
|
|
scan_duration_seconds=data.get("scan_duration_seconds") or 0.0,
|
|
analyzers_used=data.get("analyzers_used") or [],
|
|
analyzers_failed=data.get("analyzers_failed") or [],
|
|
analyzability_score=data.get("analyzability_score"),
|
|
analyzability_details=data.get("analyzability_details"),
|
|
scan_metadata=data.get("scan_metadata"),
|
|
)
|
|
|
|
|
|
def generate_json_report(report, skill_meta: dict | None = None,
|
|
run_meta: dict | None = None) -> dict:
|
|
"""Build the machine-readable companion to the markdown report.
|
|
|
|
The per-skill `scan_metadata` and `analyzability_details` are included
|
|
because they carry the scanner's own view of each package, which is what
|
|
lets a reviewer diagnose a finding that disagrees with the filesystem.
|
|
|
|
Per-skill `content_hash` and `last_scanned` are what make incremental runs
|
|
possible and honest: the hash decides whether a later run may reuse these
|
|
findings, and the timestamp records when the finding was actually produced
|
|
rather than when the report was written.
|
|
"""
|
|
skill_meta = skill_meta or {}
|
|
return {
|
|
"generated": datetime.now(timezone.utc).isoformat(timespec="seconds"),
|
|
"run": run_meta or {},
|
|
"totals": {
|
|
"skills_scanned": report.total_skills_scanned,
|
|
"findings": report.total_findings,
|
|
"critical": report.critical_count,
|
|
"high": report.high_count,
|
|
"medium": report.medium_count,
|
|
"low": report.low_count,
|
|
"info": report.info_count,
|
|
"safe_skills": report.safe_count,
|
|
},
|
|
"skills_skipped": report.skills_skipped,
|
|
"cross_skill_findings": [_finding_dict(f) for f in report.cross_skill_findings],
|
|
"skills": [
|
|
{
|
|
"name": result.skill_name,
|
|
"directory": result.skill_directory,
|
|
"is_safe": result.is_safe,
|
|
"max_severity": _enum_value(result.max_severity),
|
|
"scan_duration_seconds": round(result.scan_duration_seconds, 2),
|
|
"content_hash": skill_meta.get(result.skill_name, {}).get("content_hash"),
|
|
"last_scanned": skill_meta.get(result.skill_name, {}).get("last_scanned"),
|
|
"reused_from_previous_report": skill_meta.get(result.skill_name, {}).get(
|
|
"reused", False
|
|
),
|
|
"analyzers_used": result.analyzers_used,
|
|
"analyzers_failed": result.analyzers_failed,
|
|
"analyzability_score": result.analyzability_score,
|
|
"scan_metadata": result.scan_metadata,
|
|
"analyzability_details": result.analyzability_details,
|
|
"findings": [_finding_dict(f) for f in result.findings],
|
|
}
|
|
# Sorted so the committed JSON diffs cleanly regardless of the order
|
|
# concurrent workers happen to finish in.
|
|
for result in sorted(report.scan_results, key=lambda r: r.skill_name)
|
|
],
|
|
}
|
|
|
|
|
|
def load_previous_report(path: str | None = None) -> dict | None:
|
|
"""Load the last published JSON report, if it is usable as a cache."""
|
|
p = Path(path or JSON_OUTPUT_FILE)
|
|
if not p.exists():
|
|
return None
|
|
try:
|
|
return json.loads(p.read_text())
|
|
except (json.JSONDecodeError, OSError) as e:
|
|
print(f" Previous report unusable ({e}); scanning everything.")
|
|
return None
|
|
|
|
|
|
def cache_invalidation_reason(previous: dict | None, max_age_days: int) -> str | None:
|
|
"""Explain why the cache cannot be used, or None if it can.
|
|
|
|
Findings depend on the scanner and the model as much as on the skill, so a
|
|
change to either invalidates every cached result. The age limit is a backstop
|
|
against a cache that quietly persists for months.
|
|
"""
|
|
if previous is None:
|
|
return "no previous report"
|
|
|
|
run = previous.get("run") or {}
|
|
if run.get("scanner_version") != scanner_version():
|
|
return (
|
|
f"scanner version changed "
|
|
f"({run.get('scanner_version')} -> {scanner_version()})"
|
|
)
|
|
if run.get("model") != llm_model_id():
|
|
return f"model changed ({run.get('model')} -> {llm_model_id()})"
|
|
|
|
stamp = previous.get("run", {}).get("last_full_scan") or previous.get("generated")
|
|
if not stamp:
|
|
return "previous report has no timestamp"
|
|
try:
|
|
when = datetime.fromisoformat(stamp)
|
|
except ValueError:
|
|
return f"unparseable timestamp ({stamp!r})"
|
|
if when.tzinfo is None:
|
|
when = when.replace(tzinfo=timezone.utc)
|
|
age = datetime.now(timezone.utc) - when
|
|
if age > timedelta(days=max_age_days):
|
|
return f"last full scan was {age.days} days ago (limit {max_age_days})"
|
|
|
|
return None
|
|
|
|
|
|
def scan_with_progress(
|
|
scanner: SkillScanner,
|
|
skills_dir: str,
|
|
*,
|
|
workers: int = DEFAULT_WORKERS,
|
|
previous: dict | None = None,
|
|
) -> tuple[Report, dict, dict]:
|
|
"""Scan every skill concurrently, reusing unchanged results where possible.
|
|
|
|
Returns the report, per-skill metadata for the JSON companion, and run
|
|
metadata. A scanner is built per worker thread rather than shared: the
|
|
analyzers carry mutable per-scan state, and sharing one instance across
|
|
threads would interleave it.
|
|
"""
|
|
base = Path(skills_dir)
|
|
if not base.exists():
|
|
raise FileNotFoundError(f"Directory does not exist: {base}")
|
|
|
|
skill_dirs = sorted(
|
|
{p.parent for p in base.rglob("SKILL.md")},
|
|
key=lambda p: p.name,
|
|
)
|
|
total = len(skill_dirs)
|
|
if total == 0:
|
|
print(" No skills found.")
|
|
return Report(), {}, {}
|
|
|
|
cached: dict[str, dict] = {}
|
|
if previous is not None:
|
|
cached = {
|
|
s["name"]: s
|
|
for s in previous.get("skills") or []
|
|
if s.get("content_hash")
|
|
}
|
|
|
|
hashes = {d.name: content_hash(d) for d in skill_dirs}
|
|
fresh_dirs, reuse_dirs = [], []
|
|
for d in skill_dirs:
|
|
prior = cached.get(d.name)
|
|
if prior and prior.get("content_hash") == hashes[d.name]:
|
|
reuse_dirs.append(d)
|
|
else:
|
|
fresh_dirs.append(d)
|
|
|
|
print(
|
|
f" {total} skills: {len(fresh_dirs)} to scan, "
|
|
f"{len(reuse_dirs)} unchanged (reusing previous findings)"
|
|
)
|
|
print(f" Concurrency: {workers} workers\n")
|
|
|
|
report = Report()
|
|
skill_meta: dict[str, dict] = {}
|
|
loaded_skills = []
|
|
scan_start = time.time()
|
|
now_stamp = datetime.now(timezone.utc).isoformat(timespec="seconds")
|
|
|
|
for d in reuse_dirs:
|
|
prior = cached[d.name]
|
|
report.add_scan_result(_result_from_dict(prior))
|
|
skill_meta[d.name] = {
|
|
"content_hash": hashes[d.name],
|
|
"last_scanned": prior.get("last_scanned") or prior.get("generated"),
|
|
"reused": True,
|
|
}
|
|
|
|
# The loader is cheap and LLM-free; skills are loaded up front so the
|
|
# cross-skill overlap pass has every package, including reused ones.
|
|
for d in skill_dirs:
|
|
try:
|
|
loaded_skills.append(scanner.loader.load_skill(d))
|
|
except Exception: # counted per skill below when it is scanned
|
|
pass
|
|
|
|
local = threading.local()
|
|
|
|
def worker_scanner() -> SkillScanner:
|
|
if not hasattr(local, "scanner"):
|
|
local.scanner = build_scanner()
|
|
return local.scanner
|
|
|
|
def scan_one(skill_dir: Path) -> tuple[Path, ScanResult | None, str | None, float]:
|
|
t0 = time.time()
|
|
try:
|
|
s = worker_scanner()
|
|
skill = s.loader.load_skill(skill_dir)
|
|
return skill_dir, s._scan_single_skill(skill, skill_dir), None, time.time() - t0
|
|
except SkillLoadError as e:
|
|
return skill_dir, None, f"SKIP ({e})", time.time() - t0
|
|
except Exception as e:
|
|
return skill_dir, None, f"ERROR ({e})", time.time() - t0
|
|
|
|
longest_name = max((len(d.name) for d in skill_dirs), default=10)
|
|
width = len(str(total))
|
|
done = len(reuse_dirs)
|
|
print_lock = threading.Lock()
|
|
|
|
if fresh_dirs:
|
|
with ThreadPoolExecutor(max_workers=workers) as pool:
|
|
futures = {pool.submit(scan_one, d): d for d in fresh_dirs}
|
|
for future in as_completed(futures):
|
|
skill_dir, result, error, elapsed = future.result()
|
|
name = skill_dir.name
|
|
done += 1
|
|
counter = f"[{done:>{width}}/{total}]"
|
|
|
|
if result is not None:
|
|
report.add_scan_result(result)
|
|
skill_meta[name] = {
|
|
"content_hash": hashes[name],
|
|
"last_scanned": now_stamp,
|
|
"reused": False,
|
|
}
|
|
tag = severity_badge(_enum_value(result.max_severity))
|
|
n = len(result.findings)
|
|
detail = f"{n} finding{'s' if n != 1 else ''}" if n else ""
|
|
line = f" {counter} {name:{longest_name}} {tag:18} {detail:20} ({elapsed:.1f}s)"
|
|
else:
|
|
report.skills_skipped.append(
|
|
{"skill": str(skill_dir), "reason": error or "unknown"}
|
|
)
|
|
icon = "⚠️ " if str(error).startswith("SKIP") else "❌"
|
|
line = f" {counter} {name:{longest_name}} {icon} {error} ({elapsed:.1f}s)"
|
|
|
|
with print_lock:
|
|
print(line, flush=True)
|
|
|
|
wall = time.time() - scan_start
|
|
|
|
run_meta = {
|
|
"scanner_version": scanner_version(),
|
|
"model": llm_model_id(),
|
|
"workers": workers,
|
|
"skills_rescanned": len(fresh_dirs),
|
|
"skills_reused": len(reuse_dirs),
|
|
"wall_seconds": round(wall, 1),
|
|
}
|
|
|
|
if len(loaded_skills) > 1:
|
|
print("\n Running cross-skill overlap analysis ...", end="", flush=True)
|
|
t0 = time.time()
|
|
try:
|
|
overlap = scanner._check_description_overlap(loaded_skills)
|
|
|
|
from skill_scanner.core.analyzers.cross_skill_scanner import CrossSkillScanner
|
|
|
|
cross = CrossSkillScanner().analyze_skill_set(loaded_skills)
|
|
all_cross = [*list(overlap or []), *list(cross or [])]
|
|
if scanner.policy.disabled_rules:
|
|
all_cross = [f for f in all_cross if f.rule_id not in scanner.policy.disabled_rules]
|
|
if all_cross:
|
|
scanner._apply_severity_overrides(all_cross)
|
|
report.add_cross_skill_findings(all_cross)
|
|
elapsed = time.time() - t0
|
|
print(f" {len(all_cross)} finding{'s' if len(all_cross) != 1 else ''} ({elapsed:.1f}s)")
|
|
except Exception as e:
|
|
print(f" error: {e}")
|
|
|
|
print(f"\n Done in {wall:.1f}s")
|
|
return report, skill_meta, run_meta
|
|
|
|
|
|
def parse_args(argv=None) -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
|
|
parser.add_argument(
|
|
"--full",
|
|
action="store_true",
|
|
default=os.getenv("SKILL_SCAN_FULL") == "1",
|
|
help="rescan every skill, ignoring the previous report's cached findings",
|
|
)
|
|
parser.add_argument(
|
|
"--workers",
|
|
type=int,
|
|
default=int(os.getenv("SKILL_SCAN_WORKERS", DEFAULT_WORKERS)),
|
|
help=f"skills scanned concurrently (default: {DEFAULT_WORKERS})",
|
|
)
|
|
parser.add_argument(
|
|
"--max-age-days",
|
|
type=int,
|
|
default=int(os.getenv("SKILL_SCAN_MAX_AGE_DAYS", DEFAULT_MAX_AGE_DAYS)),
|
|
help=f"force a full rescan when the last one is older (default: {DEFAULT_MAX_AGE_DAYS})",
|
|
)
|
|
return parser.parse_args(argv)
|
|
|
|
|
|
def main(argv=None):
|
|
args = parse_args(argv)
|
|
workers = max(1, args.workers)
|
|
|
|
print("Building scanner (LLM + behavioral + trigger + balanced policy)...")
|
|
scanner = build_scanner()
|
|
print(f"Analyzers: {scanner.list_analyzers()}")
|
|
print(f"Scanner {SCANNER_PACKAGE} {scanner_version()}, model {llm_model_id()}\n")
|
|
|
|
previous = None if args.full else load_previous_report()
|
|
if args.full:
|
|
print("Full rescan requested; ignoring cached findings.")
|
|
reason = "--full requested"
|
|
else:
|
|
reason = cache_invalidation_reason(previous, args.max_age_days)
|
|
if reason:
|
|
print(f"Cache not usable: {reason}. Scanning everything.")
|
|
previous = None
|
|
|
|
print(f"Scanning {SKILLS_DIR}/...")
|
|
report, skill_meta, run_meta = scan_with_progress(
|
|
scanner, SKILLS_DIR, workers=workers, previous=previous
|
|
)
|
|
|
|
was_full = previous is None
|
|
run_meta["full_scan"] = was_full
|
|
run_meta["cache_invalidated_because"] = reason
|
|
if was_full:
|
|
run_meta["last_full_scan"] = datetime.now(timezone.utc).isoformat(timespec="seconds")
|
|
else:
|
|
# Carried forward so the age backstop measures from the last *full* scan,
|
|
# not from the last incremental one.
|
|
run_meta["last_full_scan"] = (previous.get("run") or {}).get("last_full_scan")
|
|
|
|
print(f"\nResults: {report.total_skills_scanned} skills, {report.total_findings} findings")
|
|
print(f" Critical: {report.critical_count} High: {report.high_count} Safe: {report.safe_count}")
|
|
print(
|
|
f" {run_meta['skills_rescanned']} rescanned, {run_meta['skills_reused']} reused "
|
|
f"in {run_meta['wall_seconds']}s"
|
|
)
|
|
|
|
md = generate_report(report, run_meta=run_meta)
|
|
Path(OUTPUT_FILE).parent.mkdir(parents=True, exist_ok=True)
|
|
with open(OUTPUT_FILE, "w") as f:
|
|
f.write(md)
|
|
|
|
with open(JSON_OUTPUT_FILE, "w") as f:
|
|
json.dump(
|
|
generate_json_report(report, skill_meta=skill_meta, run_meta=run_meta),
|
|
f, indent=2, sort_keys=False,
|
|
)
|
|
f.write("\n")
|
|
|
|
print(f"\nReport written to {OUTPUT_FILE}")
|
|
print(f"Machine-readable report written to {JSON_OUTPUT_FILE}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|