wehub-resource-sync
2114ccd278
CI / Lint & Test (Python 3.13) (push) Failing after 2s
CI / Lint & Test (Python 3.14) (push) Failing after 1s
CI / Lint & Test (Python 3.12) (push) Failing after 2s
CI / DCO Check (push) Has been skipped
Scorecard supply-chain security / Scorecard analysis (push) Failing after 2s
469 lines
16 KiB
Python
469 lines
16 KiB
Python
# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
|
|
"""Batch scanner for SkillSpector with multilingual enhancement and concurrent execution.
|
|
|
|
Scans a directory of AI agent skills in parallel (configurable worker pool)
|
|
and produces a single aggregated report (terminal / JSON / Markdown). For
|
|
non-English skills, runs a targeted LLM gap-fill pass covering 8 vulnerability
|
|
categories that have no semantic-analyzer equivalent.
|
|
|
|
Concurrency model
|
|
-----------------
|
|
Each skill runs the full ``graph.invoke(state)`` pipeline in a dedicated
|
|
thread via :class:`~concurrent.futures.ThreadPoolExecutor`. The number of
|
|
parallel workers is controlled by ``--workers`` (default 4). A 90-second
|
|
per-skill timeout prevents stalled workers from blocking the batch. This
|
|
sits on top of two built-in parallelism layers:
|
|
|
|
* **Layer 1** — 20 analyzers fan-out inside the LangGraph (per-skill)
|
|
* **Layer 2** — :meth:`~skillspector.llm_analyzer_base.LLMAnalyzerBase.arun_batches`
|
|
with ``Semaphore(10)`` (per-analyzer)
|
|
* **Layer 3** — ``ThreadPoolExecutor(max_workers)`` across skills (this module)
|
|
|
|
API rate-limit protection is provided by the :class:`~.api_pool.ApiKeyPool`
|
|
for **all** LLM calls — graph-internal analyzers, meta-analyzer, and gap-fill
|
|
alike. The pool is wired in via :func:`~.runner.set_api_pool` (monkey-patches
|
|
:func:`~skillspector.llm_utils.get_chat_model`) before any scan work starts.
|
|
|
|
Usage::
|
|
|
|
python -m contrib.batch_scan.batch_scan ./skills/ --no-llm
|
|
python -m contrib.batch_scan.batch_scan ./skills/ -f json -o report.json
|
|
python -m contrib.batch_scan.batch_scan ./skills/ --lang zh --workers 8
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
# -- .env must load BEFORE any skillspector imports, because constants.py
|
|
# reads SKILLSPECTOR_MODEL / SKILLSPECTOR_PROVIDER at import time.
|
|
try:
|
|
import dotenv as _dotenv # noqa: I001
|
|
except ImportError:
|
|
pass
|
|
else:
|
|
_dotenv.load_dotenv(_dotenv.find_dotenv(usecwd=True), override=True)
|
|
|
|
import argparse
|
|
import sys
|
|
import threading
|
|
from concurrent.futures import ThreadPoolExecutor, TimeoutError, as_completed
|
|
from pathlib import Path
|
|
from skillspector.constants import MODEL_CONFIG
|
|
from skillspector.logging_config import set_level
|
|
|
|
from .annotation import annotate_findings
|
|
from .api_pool import create_api_key_pool_from_env
|
|
from .detection import detect_skill_language
|
|
from .discovery import discover_skills
|
|
from .gap_fill import run_gap_fill
|
|
from .reports import _format_json as format_json
|
|
from .reports import _format_markdown as format_markdown
|
|
from .reports import _format_terminal as format_terminal
|
|
from .runner import run_one
|
|
|
|
# Directories skipped during file reads (same set as build_context._SKIP_DIRS).
|
|
_SKIP_DIRS: frozenset[str] = frozenset(
|
|
{".git", "__pycache__", "node_modules", ".venv", "venv", ".tox", ".pytest_cache"}
|
|
)
|
|
|
|
# Progress-print lock — Rich consoles are not thread-safe; serialize output
|
|
# from the main thread via this lock.
|
|
_print_lock = threading.Lock()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _read_skill_files(skill_dir: Path) -> dict[str, str]:
|
|
"""Lightweight file read for language detection and gap-fill.
|
|
|
|
Mirrors the file-walk rules in
|
|
:func:`skillspector.nodes.build_context._walk_skill_files`.
|
|
"""
|
|
file_cache: dict[str, str] = {}
|
|
for item in skill_dir.rglob("*"):
|
|
if not item.is_file():
|
|
continue
|
|
if any(skip in item.parts for skip in _SKIP_DIRS):
|
|
continue
|
|
if item.name.startswith(".") and not item.name.startswith(".claude"):
|
|
continue
|
|
try:
|
|
file_cache[str(item.relative_to(skill_dir))] = item.read_text(
|
|
encoding="utf-8", errors="replace"
|
|
)
|
|
except OSError:
|
|
continue
|
|
return file_cache
|
|
|
|
|
|
def _resolve_language(skill_dir: Path, cli_lang: str) -> str:
|
|
"""Determine the language for a skill directory.
|
|
|
|
When *cli_lang* is ``"auto"``, reads files and runs heuristic
|
|
detection. Otherwise returns *cli_lang* as-is.
|
|
"""
|
|
if cli_lang != "auto":
|
|
return cli_lang
|
|
fc = _read_skill_files(skill_dir)
|
|
if not fc:
|
|
return "en"
|
|
return detect_skill_language(fc)
|
|
|
|
|
|
def _scan_skill(
|
|
skill_dir: Path,
|
|
root: Path,
|
|
*,
|
|
use_llm: bool,
|
|
lang: str,
|
|
require_llm: bool,
|
|
api_pool=None,
|
|
) -> tuple[dict[str, object], str | None, str]:
|
|
"""Scan a single skill through the full pipeline.
|
|
|
|
Returns
|
|
-------
|
|
(entry, error_message_or_None, relative_name)
|
|
"""
|
|
try:
|
|
rel_name = str(skill_dir.relative_to(root))
|
|
except ValueError:
|
|
rel_name = skill_dir.name
|
|
|
|
# Core scan via the LangGraph graph
|
|
entry, error_msg = run_one(
|
|
skill_dir,
|
|
root,
|
|
use_llm=use_llm,
|
|
detected_language=lang,
|
|
)
|
|
|
|
# Gap-fill for non-English skills (post-graph, appends to issues)
|
|
if lang != "en" and use_llm and not error_msg:
|
|
fc = _read_skill_files(skill_dir)
|
|
gap_findings = run_gap_fill(
|
|
fc, lang, model=MODEL_CONFIG.get("default"), api_pool=api_pool
|
|
)
|
|
if gap_findings:
|
|
existing = list(entry.get("issues", []))
|
|
new_issues = annotate_findings(
|
|
[f.to_dict() for f in gap_findings], lang
|
|
)
|
|
entry["issues"] = existing + new_issues # type: ignore[operator]
|
|
# Patch enhancements so reports can show what was applied
|
|
entry["enhancements"]["gap_fill_applied"] = True
|
|
entry["enhancements"]["gap_fill_findings"] = len(gap_findings)
|
|
|
|
return entry, error_msg, rel_name
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Main
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def main() -> None:
|
|
"""Entry point for the batch scanner CLI."""
|
|
# -- DeepSeek compatibility patches (scoped context manager) --------------
|
|
# Patches are active for the entire scan and restored on exit — even if
|
|
# an exception occurs. Pattern: Save → Patch → Yield → Restore (finally).
|
|
from .runner import deepseek_compat
|
|
|
|
with deepseek_compat():
|
|
_main_impl()
|
|
|
|
|
|
def _main_impl() -> None:
|
|
"""Body of main(), wrapped by deepseek_compat context manager."""
|
|
# -- Windows Unicode support ---------------------------------------------
|
|
if sys.platform == "win32":
|
|
sys.stdout.reconfigure(encoding="utf-8", errors="replace") # type: ignore[attr-defined]
|
|
|
|
# -- Rich detection -------------------------------------------------------
|
|
try:
|
|
from rich.console import Console
|
|
except ImportError:
|
|
Console = None # type: ignore[assignment] # noqa: N806
|
|
|
|
c = Console() if Console is not None else None
|
|
|
|
def _print(*args: object, **kwargs: object) -> None:
|
|
"""Print through Rich when available, falling back to plain text."""
|
|
if c:
|
|
c.print(*args, **{k: v for k, v in kwargs.items() if k != "file"})
|
|
else:
|
|
msg = " ".join(str(a) for a in args)
|
|
file = kwargs.get("file")
|
|
if file:
|
|
print(msg, file=file) # type: ignore[arg-type]
|
|
else:
|
|
print(msg)
|
|
|
|
# -- CLI arguments -------------------------------------------------------
|
|
parser = argparse.ArgumentParser(
|
|
description="Batch-scan a directory of AI agent skills with SkillSpector.",
|
|
)
|
|
parser.add_argument(
|
|
"input_dir",
|
|
type=Path,
|
|
help="Directory containing skill subdirectories (each with a SKILL.md).",
|
|
)
|
|
parser.add_argument(
|
|
"-f",
|
|
"--format",
|
|
choices=("terminal", "json", "markdown"),
|
|
default="terminal",
|
|
help="Output format (default: terminal).",
|
|
)
|
|
parser.add_argument(
|
|
"-o",
|
|
"--output",
|
|
type=Path,
|
|
default=None,
|
|
help="Write report to FILE (default: stdout).",
|
|
)
|
|
parser.add_argument(
|
|
"--no-llm",
|
|
action="store_true",
|
|
default=False,
|
|
help="Skip LLM analysis — static patterns only.",
|
|
)
|
|
parser.add_argument(
|
|
"--workers",
|
|
type=int,
|
|
default=4,
|
|
metavar="N",
|
|
help="Number of parallel scan workers (default: 4). "
|
|
"Reduce to 1 for free-tier API keys, increase for enterprise tiers. "
|
|
"Skills that time out (90s) are skipped; other workers continue.",
|
|
)
|
|
parser.add_argument(
|
|
"-V",
|
|
"--verbose",
|
|
action="store_true",
|
|
default=False,
|
|
help="Enable DEBUG-level logging.",
|
|
)
|
|
parser.add_argument(
|
|
"--lang",
|
|
choices=("auto", "en", "zh", "ja", "ko"),
|
|
default="auto",
|
|
help="Expected skill language (default: auto-detect).",
|
|
)
|
|
parser.add_argument(
|
|
"--require-llm",
|
|
action="store_true",
|
|
default=True,
|
|
help="Require LLM for non-English skills (default).",
|
|
)
|
|
parser.add_argument(
|
|
"--no-require-llm",
|
|
action="store_false",
|
|
dest="require_llm",
|
|
help="Allow non-English scans without LLM (results will be incomplete).",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
if args.verbose:
|
|
set_level("DEBUG")
|
|
|
|
# -- Validation ----------------------------------------------------------
|
|
root = args.input_dir.resolve()
|
|
if not root.is_dir():
|
|
_print(f"[red]Error:[/red] {root} is not a directory", file=sys.stderr)
|
|
sys.exit(2)
|
|
|
|
skill_dirs = discover_skills(root)
|
|
if not skill_dirs:
|
|
_print(
|
|
"[yellow]No skills found.[/yellow] Each skill must be a subdirectory "
|
|
"containing a SKILL.md file.",
|
|
file=sys.stderr,
|
|
)
|
|
sys.exit(2)
|
|
|
|
# -- API Pool (optional — returns None if single-key) --------------------
|
|
api_pool = create_api_key_pool_from_env()
|
|
if api_pool:
|
|
from .runner import set_api_pool
|
|
set_api_pool(api_pool)
|
|
use_llm = not args.no_llm
|
|
|
|
# -- Header --------------------------------------------------------------
|
|
pool_note = (
|
|
f", [green]{api_pool.keys_configured} keys "
|
|
f"({api_pool.total_capacity} slots)[/green]"
|
|
if api_pool
|
|
else ""
|
|
)
|
|
_print(
|
|
f"\n[bold]SkillSpector Batch Scan[/bold] — "
|
|
f"{len(skill_dirs)} skill(s) in [dim]{root}[/dim]"
|
|
f" ([cyan]{args.workers} workers[/cyan]{pool_note})\n"
|
|
)
|
|
|
|
# -- Scan (parallel) -----------------------------------------------------
|
|
results: list[dict[str, object]] = []
|
|
errors = 0
|
|
has_high_risk = False
|
|
|
|
_sev_colors: dict[str, str] = {
|
|
"LOW": "green",
|
|
"MEDIUM": "yellow",
|
|
"HIGH": "red",
|
|
"CRITICAL": "bold red",
|
|
"ERROR": "red",
|
|
}
|
|
|
|
# Pre-resolve languages so worker threads don't contend on file I/O
|
|
lang_map: dict[Path, str] = {}
|
|
for skill_dir in skill_dirs:
|
|
lang_map[skill_dir] = _resolve_language(skill_dir, args.lang)
|
|
|
|
total = len(skill_dirs)
|
|
|
|
with ThreadPoolExecutor(max_workers=args.workers) as executor:
|
|
future_map = {
|
|
executor.submit(
|
|
_scan_skill,
|
|
skill_dir,
|
|
root,
|
|
use_llm=use_llm,
|
|
lang=lang_map[skill_dir],
|
|
require_llm=args.require_llm,
|
|
api_pool=api_pool,
|
|
): idx
|
|
for idx, skill_dir in enumerate(skill_dirs, 1)
|
|
}
|
|
|
|
for future in as_completed(future_map):
|
|
idx = future_map[future]
|
|
rel_name = str(skill_dirs[idx - 1].relative_to(root)) if idx <= len(skill_dirs) else "?"
|
|
try:
|
|
entry, error_msg, rel_name = future.result(timeout=90)
|
|
except TimeoutError:
|
|
errors += 1
|
|
with _print_lock:
|
|
_print(
|
|
f" [{idx}/{total}] [cyan]{rel_name}[/cyan] → "
|
|
f"[red]TIMEOUT (90s)[/red]"
|
|
)
|
|
# Don't retry — the worker thread is still stuck and a
|
|
# retry would consume another slot. HTTP-level timeouts
|
|
# (runner.py Patch 6) prevent most hangs from happening.
|
|
continue
|
|
except Exception:
|
|
# Unexpected crash (e.g. asyncio event-loop failure).
|
|
# Don't retry — log and continue.
|
|
errors += 1
|
|
with _print_lock:
|
|
_print(
|
|
f" [{idx}/{total}] [cyan]{rel_name}[/cyan] → "
|
|
f"[red]CRASH[/red]"
|
|
)
|
|
continue
|
|
lang = lang_map[skill_dirs[idx - 1]]
|
|
results.append(entry)
|
|
|
|
# -- Progress (main thread via lock — safe for Rich) ---------
|
|
with _print_lock:
|
|
# Non-English LLM guard warning
|
|
if lang != "en" and not use_llm and args.require_llm:
|
|
_print(
|
|
f"[yellow]WARNING:[/yellow] non-English skill "
|
|
f"'{rel_name}' ({lang}) scanned with --no-llm. "
|
|
f"Static pattern recall is reduced for this language. "
|
|
f"Re-run without --no-llm for full coverage, or use "
|
|
f"--no-require-llm to suppress this warning.",
|
|
file=sys.stderr,
|
|
)
|
|
|
|
if error_msg:
|
|
errors += 1
|
|
_print(
|
|
f" [{idx}/{total}] [cyan]{rel_name}[/cyan] → "
|
|
f"[red]ERROR: {error_msg}[/red]"
|
|
)
|
|
else:
|
|
risk = entry.get("risk_assessment", {})
|
|
score = risk.get("score", 0)
|
|
severity = risk.get("severity", "LOW")
|
|
n_issues = len(entry.get("issues", []))
|
|
if score > 50:
|
|
has_high_risk = True
|
|
color = _sev_colors.get(severity, "")
|
|
_print(
|
|
f" [{idx}/{total}] [cyan]{rel_name}[/cyan] → "
|
|
f"[{color}]{score}/100 {severity}[/{color}] "
|
|
f"({n_issues} issue(s))"
|
|
)
|
|
|
|
# -- Sort results by risk score descending -------------------------------
|
|
results.sort(
|
|
key=lambda x: x.get("risk_assessment", {}).get("score", 0), # type: ignore[no-any-return]
|
|
reverse=True,
|
|
)
|
|
|
|
# -- API Pool summary (if active) ----------------------------------------
|
|
if api_pool:
|
|
snap = api_pool.snapshot()
|
|
_parts = [
|
|
f"{snap['total_requests_served']} requests served",
|
|
]
|
|
if snap.get("peak_active_requests", 0) > 0:
|
|
_parts.append(
|
|
f"peak {snap['peak_active_requests']}/{snap['total_capacity']} slots"
|
|
)
|
|
if snap.get("rate_limits_hit", 0) > 0:
|
|
_parts.append(
|
|
f"{snap['rate_limits_hit']} rate-limit(s), "
|
|
f"{snap['retry_successes']} retried"
|
|
)
|
|
_parts.append(f"{snap['keys_configured']} keys")
|
|
_print(f"\n[dim]API Pool: {', '.join(_parts)}[/dim]")
|
|
|
|
# -- Output --------------------------------------------------------------
|
|
fmt = args.format
|
|
if fmt == "terminal":
|
|
report_body = format_terminal(results)
|
|
elif fmt == "json":
|
|
report_body = format_json(results)
|
|
else:
|
|
report_body = format_markdown(results)
|
|
|
|
if args.output:
|
|
args.output.write_text(report_body, encoding="utf-8")
|
|
_print(f"\n[green]Batch report saved to:[/green] {args.output}")
|
|
else:
|
|
if fmt == "terminal":
|
|
_print(report_body)
|
|
else:
|
|
sys.stdout.write(report_body + "\n")
|
|
|
|
# -- Exit codes ----------------------------------------------------------
|
|
if errors:
|
|
sys.exit(2)
|
|
if has_high_risk:
|
|
sys.exit(1)
|
|
# else: exit 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|