claude-plugins-official/plugins/claude-security/scripts/write_scan_meta.py

#!/usr/bin/env python3
"""Write scan-meta.json for a run: the record of what was scanned.

Mints the scan's id and captures, from git itself: the revision, the scan
root's path within the repository, the credential-free https form of its
remote 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 with any root-level symbolic links left out of them.

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, 1 on a refusal naming what is wrong, 2 on a usage error;
the file is written only on success.
Python 3.9-compatible, stdlib only.
"""

from __future__ import annotations

import argparse
import os
import re
import stat
import subprocess
import sys
import uuid
from pathlib import Path
from typing import Literal, NamedTuple, TypedDict
from urllib.parse import quote, unquote, urlsplit

# The lib/ package lives next to this script. Python normally adds a script's own
# directory to the import path, but not under -P or PYTHONSAFEPATH, so we add it here.
sys.path.insert(0, str(Path(__file__).resolve().parent))
from lib import absolute, console, plugin, strictjson

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
    sparse: Literal[True]
    not_checked_out_dirs: list[str]
    base: str | None
    merge_base: str | None


class Args(argparse.Namespace):
    """The parsed command line."""

    run_dir: str = ""
    scan_root: str = ""
    mode: str = ""
    effort: str = ""
    scope: str = ""
    base: str | None = None
    merge_base: str | None = None
    commit: str | None = None


class MetaError(Exception):
    """A refusal: the command line was well-formed but the run cannot be recorded."""


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", "surrogateescape").rstrip("\r\n")


class Extent(NamedTuple):
    """The scan target's top-level directories, its root-level symbolic links, and
    the tracked top-level directories its working tree does not hold."""

    dirs: list[str]
    symlinks: list[str]
    absent: list[str]


def tree_extent(scan_root: str) -> Extent | 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. Entries are classified without following
    symbolic links, so nothing outside the checkout is read: a root-level
    symbolic link is never one of the directories, and is named in `symlinks`
    so the report can say it was not followed. `.git` and `CLAUDE-SECURITY-*`
    report directories are excluded. None when the tree could not be listed.
    """
    names: set[str] = set()
    symlinks: 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:
                try:
                    mode = os.lstat(os.path.join(scan_root, path)).st_mode
                except OSError:
                    continue
                if stat.S_ISLNK(mode):
                    symlinks.add(path)
                elif stat.S_ISDIR(mode):
                    names.add(path)
    else:
        try:
            with os.scandir(scan_root) as entries:
                for entry in entries:
                    if entry.is_symlink():
                        symlinks.add(entry.name)
                    elif entry.is_dir(follow_symlinks=False):
                        names.add(entry.name)
        except OSError:
            return None
        names.discard(".git")
    kept = sorted(n for n in names if not n.startswith(plugin.REPORT_DIR_PREFIX))
    on_disk = {n for n in kept if os.path.lexists(os.path.join(scan_root, n))}
    dirs = [n for n in kept if n in on_disk]
    return Extent(dirs, sorted(symlinks), [n for n in kept if n not in on_disk])


def sparse_checkout(scan_root: str, extent: Extent | None) -> list[str] | None:
    """The tracked top-level directories a sparse checkout left out; None when it is not one."""
    if git(scan_root, "config", "--bool", "core.sparseCheckout") != "true":
        return None
    return extent.absent if extent else []


REMOTE_SCHEMES = frozenset({"http", "https", "ssh", "git", "git+ssh"})


def sanitize_remote(url: str | None) -> str | None:
    """`url` as a credential-free https URL naming the same repository, or None.

    Userinfo, query and fragment are stripped; the scheme becomes https and the
    host lowercase; a port survives only from an http or https URL; scp-like
    `user@host:path` is read as ssh; a trailing `/` or `.git` is dropped, so
    the ssh and https spellings of one repository come out equal. A URL that
    does not name a hosted repository is None.
    """
    text = (url or "").strip()
    if not text or "[" in text or "]" in text or strictjson.has_lone_surrogate(text):
        return None
    if "://" in text:
        try:
            parts = urlsplit(text)
            port = parts.port if parts.scheme.lower() in {"http", "https"} else None
        except ValueError:
            return None
        if parts.scheme.lower() not in REMOTE_SCHEMES:
            return None
        host = (parts.hostname or "").lower()
        location = host if port is None else f"{host}:{port}"
        path = parts.path
    else:
        # Userinfo splits off first: an optional user@ group backtracks and leaks the secret.
        rest = text.rpartition("@")[2]
        matched = re.match(r"([^@:/\\]{2,}):(.*)", rest)
        if not matched:
            return None
        location = matched[1].lower()
        path = matched[2]
    if not re.fullmatch(r"[a-z0-9.-]+(?::\d+)?", location):
        return None
    path = quote(unquote(path.strip("/"))).removesuffix(".git").rstrip("/")
    if not path:
        return None
    return f"https://{location}/{path}"


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]
        if any(part.startswith(plugin.REPORT_DIR_PREFIX) for part in path.split("/")[:-1]):
            continue
        return True
    return False


def capture_revision(scan_root: str, opts: Args) -> 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 scoped(entry: str, scan_root: str) -> str:
    """A scope entry relative to the scan root; an absolute one that is not inside it is refused."""
    if not absolute.spelled(entry):
        return entry
    literal = os.path.abspath(entry)
    parent, name = os.path.split(literal)
    try:
        resolutions = [
            literal,
            os.path.join(os.path.realpath(parent), name),
            os.path.realpath(entry),
        ]
    except OSError:
        resolutions = [literal]
    for resolved in resolutions:
        try:
            relative = os.path.relpath(resolved, scan_root).replace("\\", "/")
        except ValueError:
            continue
        if relative != ".." and not relative.startswith("../"):
            return relative
    msg = f"--scope entry {entry!r} is not inside the scan root {scan_root!r}"
    raise MetaError(msg)


def parse_options(argv: list[str]) -> Args:
    """The parsed command line; anything wrong with it is argparse's exit 2."""
    ap = argparse.ArgumentParser(prog="write_scan_meta")
    ap.add_argument("run_dir")
    ap.add_argument("scan_root")
    ap.add_argument("--mode", required=True, choices=plugin.MODES)
    ap.add_argument("--effort", required=True, choices=["low", "medium", "high", "max"])
    ap.add_argument("--scope")
    ap.add_argument("--base")
    ap.add_argument("--merge-base", dest="merge_base")
    ap.add_argument("--commit")
    opts = ap.parse_args(argv, namespace=Args())
    if opts.mode == "commit" and not opts.commit:
        ap.error("--mode commit requires --commit <sha>")
    if not os.path.isdir(opts.run_dir):
        ap.error(f"run directory does not exist: {opts.run_dir}")
    return opts


def main(argv: list[str]) -> int:
    opts = parse_options(argv)
    run_dir = os.path.realpath(os.path.abspath(opts.run_dir))
    scan_root = os.path.realpath(os.path.abspath(opts.scan_root))
    revision = capture_revision(scan_root, opts)
    extent = tree_extent(scan_root)
    absent = sparse_checkout(scan_root, extent) if revision.get("versioned") else None
    if absent is not None:
        revision["sparse"] = True
        revision["not_checked_out_dirs"] = absent
    scan_prefix = (
        git(scan_root, "rev-parse", "--show-prefix") if revision.get("versioned") else None
    )
    remote = (
        sanitize_remote(git(scan_root, "remote", "get-url", "origin"))
        if revision.get("versioned")
        else None
    )
    scope = [scoped(entry.strip(), scan_root) for entry in opts.scope.split(",") if entry.strip()]
    if scope and all(s in {".", "./"} for s in scope):
        scope = []
    whole_repo = opts.mode == "scan" and not scope
    if whole_repo and extent is None:
        sys.stderr.write(f"write_scan_meta: could not list {scan_root}; top_level_dirs unknown\n")
    top_level, symlinks = (extent.dirs, extent.symlinks) if whole_repo and extent else (None, None)
    if symlinks:
        sys.stderr.write(
            "write_scan_meta: root-level symbolic links not followed, "
            f"left out of top_level_dirs: {', '.join(symlinks)}\n"
        )
    meta: dict[str, object] = {
        "scan_id": str(uuid.uuid4()),
        "scan_root": scan_root,
        "scan_prefix": scan_prefix,
        "remote": remote,
        "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,
        "unfollowed_symlinks": symlinks,
    }
    path = os.path.join(run_dir, "scan-meta.json")
    with open(path, "w", encoding="utf-8", newline="\n") as out:
        out.write(strictjson.text(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")
    if absent is not None:
        listed = strictjson.text(absent)
        sys.stdout.write(f"sparse checkout: top-level directories not checked out: {listed}\n")
    sys.stdout.write(f"top_level_dirs: {strictjson.text(top_level)}\n")
    return 0


if __name__ == "__main__":
    console.tolerate_undecodable_names()
    try:
        sys.exit(main(sys.argv[1:]))
    except MetaError as error:
        sys.stderr.write(f"write_scan_meta: {error}\n")
        sys.exit(1)
    except OSError as error:
        sys.stderr.write(f"write_scan_meta: could not write the run's output: {error}\n")
        sys.exit(1)