mirror of
https://github.com/anthropics/claude-plugins-official.git
synced 2026-07-27 04:53:33 +00:00
232 lines
8.0 KiB
Python
Executable File
232 lines
8.0 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Write scan-meta.json for a run: the record of what was scanned.
|
|
|
|
Captures the revision from git itself and, for a whole-repository scan, the
|
|
tree's top-level directories, printed as a JSON array on a `top_level_dirs:`
|
|
line and recorded in the meta file.
|
|
|
|
Usage:
|
|
write_scan_meta.py <run_dir> <scan_root> --mode scan|changes|commit
|
|
--effort low|medium|high|max [--scope a,b] [--base <ref>]
|
|
[--merge-base <sha>] [--commit <sha>]
|
|
|
|
Exits 0 on success. A caller error prints a one-line diagnostic to stderr and
|
|
exits non-zero without writing the file.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
from typing import TypedDict, cast
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
from render_report import RenderError, atomic_write
|
|
|
|
PLUGIN_NAME = "claude-security"
|
|
REPORT_DIR_PREFIX = "CLAUDE-SECURITY-"
|
|
GIT_ENV = dict(os.environ, GIT_TERMINAL_PROMPT="0")
|
|
|
|
|
|
class Revision(TypedDict, total=False):
|
|
"""What was scanned. `versioned` is always present; the rest when in git."""
|
|
|
|
versioned: bool
|
|
commit: str | None
|
|
parent: str | None
|
|
branch: str | None
|
|
dirty: bool | None
|
|
base: str | None
|
|
merge_base: str | None
|
|
|
|
|
|
class Options(TypedDict):
|
|
"""The parsed, typed command line -- argparse hands back untyped attributes."""
|
|
|
|
run_dir: str
|
|
scan_root: str
|
|
mode: str
|
|
effort: str
|
|
scope: str
|
|
base: str | None
|
|
merge_base: str | None
|
|
commit: str | None
|
|
|
|
|
|
class MetaError(Exception):
|
|
"""An input error the caller must correct."""
|
|
|
|
|
|
def _opt_str(value: object) -> str | None:
|
|
"""An argparse optional as str-or-None, typed."""
|
|
return None if value is None else str(value)
|
|
|
|
|
|
def git(cwd: str, *args: str) -> str | None:
|
|
"""One read-only git call, prompts suppressed. None on any failure."""
|
|
try:
|
|
out = subprocess.run(
|
|
["git", "-C", cwd, *args],
|
|
env=GIT_ENV,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.DEVNULL,
|
|
timeout=30,
|
|
check=False,
|
|
)
|
|
except (OSError, subprocess.SubprocessError):
|
|
return None
|
|
if out.returncode != 0:
|
|
return None
|
|
return out.stdout.decode("utf-8", "replace").rstrip("\r\n")
|
|
|
|
|
|
def top_level_dirs(scan_root: str) -> list[str] | None:
|
|
"""The scan target's top-level directories, computed from the tree itself.
|
|
|
|
Inside a git work tree the tracked files decide; where nothing is tracked
|
|
the immediate subdirectories do. `.git` and `CLAUDE-SECURITY-*` report
|
|
directories are excluded. None when the tree could not be listed.
|
|
"""
|
|
names: set[str] = set()
|
|
listing = git(scan_root, "ls-files", "-z")
|
|
if listing:
|
|
for path in listing.split("\0"):
|
|
top, sep, _rest = path.partition("/")
|
|
if sep and top:
|
|
names.add(top)
|
|
elif path and os.path.isdir(os.path.join(scan_root, path)):
|
|
names.add(path)
|
|
else:
|
|
try:
|
|
with os.scandir(scan_root) as entries:
|
|
names.update(entry.name for entry in entries if entry.is_dir(follow_symlinks=False))
|
|
except OSError:
|
|
return None
|
|
names.discard(".git")
|
|
return sorted(n for n in names if not n.startswith(REPORT_DIR_PREFIX))
|
|
|
|
|
|
def worktree_dirty(scan_root: str) -> bool | None:
|
|
"""True/False/None (unknown) for the working tree, ignoring report dirs."""
|
|
status = git(scan_root, "status", "--porcelain", "--untracked-files=all")
|
|
if status is None:
|
|
return None
|
|
for line in status.splitlines():
|
|
if len(line) < len("XY P"):
|
|
continue
|
|
path = line[3:].split(" -> ")[-1]
|
|
top = path.split("/", 1)[0]
|
|
if top.startswith(REPORT_DIR_PREFIX):
|
|
continue
|
|
return True
|
|
return False
|
|
|
|
|
|
def capture_revision(scan_root: str, opts: Options) -> Revision:
|
|
versioned = git(scan_root, "rev-parse", "--is-inside-work-tree") == "true"
|
|
if opts["mode"] == "commit":
|
|
if not versioned:
|
|
msg = f"--mode commit needs a git repository; {scan_root!r} is not one"
|
|
raise MetaError(msg)
|
|
commit_arg = opts["commit"] or ""
|
|
sha = git(scan_root, "rev-parse", "--verify", "--quiet", commit_arg + "^{commit}")
|
|
if not sha:
|
|
msg = f"--commit {commit_arg!r} does not resolve to a commit"
|
|
raise MetaError(msg)
|
|
return {
|
|
"versioned": True,
|
|
"commit": sha,
|
|
"parent": git(scan_root, "rev-parse", "--verify", "--quiet", sha + "^") or None,
|
|
"branch": git(scan_root, "rev-parse", "--abbrev-ref", "HEAD"),
|
|
"dirty": False,
|
|
}
|
|
if not versioned:
|
|
return {"versioned": False}
|
|
revision: Revision = {
|
|
"versioned": True,
|
|
"commit": git(scan_root, "rev-parse", "HEAD"),
|
|
"branch": git(scan_root, "rev-parse", "--abbrev-ref", "HEAD"),
|
|
"dirty": worktree_dirty(scan_root),
|
|
}
|
|
if opts["mode"] == "changes":
|
|
revision["base"] = opts["base"]
|
|
revision["merge_base"] = opts["merge_base"]
|
|
return revision
|
|
|
|
|
|
def parse_options(argv: list[str]) -> Options:
|
|
ap = argparse.ArgumentParser(prog="write_scan_meta")
|
|
ap.add_argument("run_dir")
|
|
ap.add_argument("scan_root")
|
|
ap.add_argument("--mode", required=True, choices=["scan", "changes", "commit"])
|
|
ap.add_argument("--effort", required=True, choices=["low", "medium", "high", "max"])
|
|
ap.add_argument("--scope", default="")
|
|
ap.add_argument("--base", default=None)
|
|
ap.add_argument("--merge-base", dest="merge_base", default=None)
|
|
ap.add_argument("--commit", default=None)
|
|
ns = ap.parse_args(argv)
|
|
return {
|
|
"run_dir": str(cast("object", ns.run_dir)),
|
|
"scan_root": str(cast("object", ns.scan_root)),
|
|
"mode": str(cast("object", ns.mode)),
|
|
"effort": str(cast("object", ns.effort)),
|
|
"scope": str(cast("object", ns.scope)),
|
|
"base": _opt_str(cast("object", ns.base)),
|
|
"merge_base": _opt_str(cast("object", ns.merge_base)),
|
|
"commit": _opt_str(cast("object", ns.commit)),
|
|
}
|
|
|
|
|
|
def main(argv: list[str]) -> int:
|
|
opts = parse_options(argv)
|
|
if opts["mode"] == "commit" and not opts["commit"]:
|
|
msg = "--mode commit requires --commit <sha>"
|
|
raise MetaError(msg)
|
|
|
|
run_dir = os.path.realpath(os.path.abspath(opts["run_dir"]))
|
|
if not os.path.isdir(run_dir):
|
|
msg = f"run directory does not exist: {run_dir}"
|
|
raise MetaError(msg)
|
|
scan_root = os.path.realpath(os.path.abspath(opts["scan_root"]))
|
|
revision = capture_revision(scan_root, opts)
|
|
scope = [s.strip() for s in opts["scope"].split(",") if s.strip()]
|
|
if scope and all(s in {".", "./"} for s in scope):
|
|
scope = []
|
|
whole_repo = opts["mode"] == "scan" and not scope
|
|
top_level = top_level_dirs(scan_root) if whole_repo else None
|
|
if whole_repo and top_level is None:
|
|
sys.stderr.write(f"write_scan_meta: could not list {scan_root}; top_level_dirs unknown\n")
|
|
meta: dict[str, object] = {
|
|
"scan_root": scan_root,
|
|
"run_dir": run_dir,
|
|
"flow": "scan" if opts["mode"] == "scan" else "changes",
|
|
"agent": f"{PLUGIN_NAME}:{PLUGIN_NAME}",
|
|
"mode": opts["mode"],
|
|
"scope": scope,
|
|
"effort": opts["effort"],
|
|
"model": None,
|
|
"revision": revision,
|
|
"revision_source": "self-reported",
|
|
"top_level_dirs": top_level,
|
|
}
|
|
path = os.path.join(run_dir, "scan-meta.json")
|
|
atomic_write(path, json.dumps(meta, indent=2) + "\n")
|
|
sys.stdout.write(f"scan-meta.json written: {path}\n")
|
|
sys.stdout.write(f"revision: {revision.get('commit') or 'UNVERSIONED'}\n")
|
|
sys.stdout.write(f"top_level_dirs: {json.dumps(top_level)}\n")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
sys.exit(main(sys.argv[1:]))
|
|
except (MetaError, RenderError) as error:
|
|
sys.stderr.write(f"write_scan_meta: {error}\n")
|
|
sys.exit(2)
|
|
except OSError as error:
|
|
sys.stderr.write(f"write_scan_meta: could not write the run's output: {error}\n")
|
|
sys.exit(2)
|