stack/dev/scripts/pkg_issues.py

"""Auto-create Gitea issues for missing packages and vulnerabilities.

Reads drift report and vulnerability scan results, creates/closes
Gitea issues via the API.

Usage:
    uv run python dev/scripts/pkg_issues.py                    # all checks
    uv run python dev/scripts/pkg_issues.py --drift-only       # missing/surplus only
    uv run python dev/scripts/pkg_issues.py --vuln-only        # CVEs only
    uv run python dev/scripts/pkg_issues.py --dry-run          # show what would happen
"""

from __future__ import annotations

import json
import os
import re
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parents[2]
MANIFEST_PATH = ROOT / "data" / "pkg-manifest.json"
# trivy JSON reports, one per lockfile (uv.lock, docs/pnpm-lock.yaml)
VULN_PATHS = (
    ROOT / "data" / "pkg-vulns.json",
    ROOT / "data" / "pkg-vulns-docs.json",
)

GITEA_URL = os.environ.get("GITEA_URL", "http://localhost:3000")
GITEA_TOKEN = os.environ.get("GITEA_TOKEN", "")
REPO = os.environ.get("GITEA_REPO", "homelab/stack")
MILESTONE_TITLE = "P21: Package Supply Chain"

# Labels to attach (by name — resolved to IDs at runtime)
LABEL_CI = "ci"
LABEL_QUALITY = "quality"


def _api(method: str, path: str, data: dict | None = None) -> dict | list | None:
    """Call Gitea API. Uses docker exec if GITEA_TOKEN not in env."""
    import subprocess

    token = GITEA_TOKEN
    if not token:
        # Try to load from .env
        env_file = ROOT / ".env"
        if env_file.exists():
            for line in env_file.read_text().splitlines():
                if line.startswith("GITEA_TOKEN="):
                    token = line.split("=", 1)[1].strip().strip('"').strip("'")
                    break

    if not token:
        print("ERROR: GITEA_TOKEN not set and not found in .env")
        raise SystemExit(2)

    url = f"http://localhost:3000/api/v1/{path}"
    cmd = [
        "docker",
        "exec",
        "git",
        "curl",
        "-s",
        "-H",
        f"Authorization: token {token}",
        "-H",
        "Content-Type: application/json",
    ]
    if method == "POST":
        cmd += ["-X", "POST", "-d", json.dumps(data), url]
    elif method == "PATCH":
        cmd += ["-X", "PATCH", "-d", json.dumps(data), url]
    else:
        cmd += [url]

    result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
    if result.returncode != 0:
        print(f"  API error: {result.stderr[:200]}")
        return None
    try:
        return json.loads(result.stdout)
    except json.JSONDecodeError:
        return None


def _get_milestone_id() -> int | None:
    """Find the P21 milestone ID."""
    milestones = _api("GET", f"repos/{REPO}/milestones?limit=50")
    if not milestones:
        return None
    for m in milestones:
        if MILESTONE_TITLE in m.get("title", ""):
            return m["id"]
    return None


def _get_label_ids(names: list[str]) -> list[int]:
    """Resolve label names to IDs."""
    labels = _api("GET", f"repos/{REPO}/labels?limit=50")
    if not labels:
        return []
    name_to_id = {lb["name"]: lb["id"] for lb in labels}
    return [name_to_id[n] for n in names if n in name_to_id]


def _find_existing_issues(prefix: str) -> dict[str, list[dict]]:
    """Find open issues with titles starting with prefix, grouped by title.

    Walks every page: a single 50-issue page hid most of the tracker once
    CI-failure issues piled up, so known CVEs were re-filed nightly (174
    issues for 91 titles) and resolved ones were never closed.
    """
    found: dict[str, list[dict]] = {}
    page = 1
    while True:
        issues = _api(
            "GET",
            f"repos/{REPO}/issues?state=open&type=issues&limit=50"
            f"&page={page}&labels={LABEL_CI}",
        )
        if not issues:
            return found
        for i in issues:
            if i["title"].startswith(prefix):
                found.setdefault(i["title"], []).append(i)
        page += 1


def _close_all(title: str, issues: list[dict], *, dry_run: bool) -> None:
    """Close every open issue carrying ``title`` (duplicates included)."""
    for issue in issues:
        if dry_run:
            print(f"  would close #{issue['number']}: {title}")
        else:
            _api(
                "PATCH",
                f"repos/{REPO}/issues/{issue['number']}",
                {"state": "closed"},
            )
            print(f"  closed #{issue['number']}: {title}")


def handle_drift(*, dry_run: bool = False) -> int:
    """Create issues for drift (missing/untracked packages)."""
    # Run drift detection inline
    sys.path.insert(0, str(ROOT / "dev" / "scripts"))
    from pkg_drift import check_drift

    drift = check_drift()
    if not drift["has_drift"]:
        print("  drift: no drift detected")
        return 0

    milestone_id = _get_milestone_id()
    label_ids = _get_label_ids([LABEL_CI])
    existing = _find_existing_issues("[pkg-missing]")
    created = 0

    for pkg_type in ("apt", "apk"):
        section = drift.get(pkg_type, {})
        for pkg in section.get("untracked", []):
            title = f"[pkg-missing] {pkg_type}/{pkg}"
            if title in existing:
                continue
            body = (
                f"Package `{pkg}` found in source ({pkg_type}) "
                f"but not in the package manifest.\n\n"
                f"Run `uv run python dev/scripts/pkg_inventory.py` to update."
            )
            if dry_run:
                print(f"  would create: {title}")
            else:
                issue_data = {"title": title, "body": body, "labels": label_ids}
                if milestone_id:
                    issue_data["milestone"] = milestone_id
                result = _api("POST", f"repos/{REPO}/issues", issue_data)
                if result:
                    print(f"  created #{result['number']}: {title}")
                    created += 1

    # PyPI sections
    pypi_drift = drift.get("pypi", {})
    for sub, d in pypi_drift.items():
        for pkg in d.get("untracked", []):
            title = f"[pkg-missing] pypi/{pkg} ({sub})"
            if title in existing:
                continue
            body = (
                f"PyPI package `{pkg}` found in source ({sub}) "
                f"but not in the package manifest.\n\n"
                f"Run `uv run python dev/scripts/pkg_inventory.py` to update."
            )
            if dry_run:
                print(f"  would create: {title}")
            else:
                issue_data = {"title": title, "body": body, "labels": label_ids}
                if milestone_id:
                    issue_data["milestone"] = milestone_id
                result = _api("POST", f"repos/{REPO}/issues", issue_data)
                if result:
                    print(f"  created #{result['number']}: {title}")
                    created += 1

    # Close issues for packages that are no longer missing
    for title, issues in existing.items():
        # Extract package name from title
        m = re.match(r"\[pkg-missing\] (\w+)/(.+?)(?:\s|$)", title)
        if not m:
            continue
        pkg_type, pkg_name = m.group(1), m.group(2)
        still_missing = False
        if pkg_type in ("apt", "apk"):
            still_missing = pkg_name in drift.get(pkg_type, {}).get("untracked", [])
        elif pkg_type == "pypi":
            for sub_d in pypi_drift.values():
                if pkg_name in sub_d.get("untracked", []):
                    still_missing = True
                    break
        issues = sorted(issues, key=lambda i: i.get("number", 0))
        _close_all(title, issues[1:] if still_missing else issues, dry_run=dry_run)

    return created


def handle_vulns(*, dry_run: bool = False) -> int:
    """Create issues for vulnerabilities found in package scans.

    Also closes issues whose CVE no longer appears — including when a scan
    comes back clean, which is exactly when the most issues resolve.
    """
    reports = [p for p in VULN_PATHS if p.exists()]
    if not reports:
        print("  vulns: no scan results found (data/pkg-vulns*.json)")
        return 0

    # trivy JSON format: {"Results": [{"Vulnerabilities": [...]}]}
    vulns: list[dict] = []
    for path in reports:
        data = json.loads(path.read_text())
        for result in data.get("Results", []):
            for v in result.get("Vulnerabilities") or []:
                vulns.append(v)

    if not vulns:
        print("  vulns: no vulnerabilities found")

    milestone_id = _get_milestone_id() if vulns else None
    label_ids = _get_label_ids([LABEL_CI, LABEL_QUALITY]) if vulns else []
    existing = _find_existing_issues("[pkg-vuln]")
    created = 0

    for v in vulns:
        cve = v.get("VulnerabilityID", "UNKNOWN")
        pkg = v.get("PkgName", "unknown")
        version = v.get("InstalledVersion", "?")
        severity = v.get("Severity", "UNKNOWN")
        fixed = v.get("FixedVersion", "none")
        desc = v.get("Description", "")[:500]
        title = f"[pkg-vuln] {cve} in {pkg}@{version}"

        if title in existing:
            continue

        body = (
            f"**Severity:** {severity}\n"
            f"**Package:** `{pkg}` @ `{version}`\n"
            f"**Fixed in:** `{fixed}`\n\n"
            f"{desc}\n\n"
            f"**Reference:** https://nvd.nist.gov/vuln/detail/{cve}"
        )

        if dry_run:
            print(f"  would create: {title}")
            existing[title] = []
        else:
            issue_data = {"title": title, "body": body, "labels": label_ids}
            if milestone_id:
                issue_data["milestone"] = milestone_id
            result = _api("POST", f"repos/{REPO}/issues", issue_data)
            if result:
                print(f"  created #{result['number']}: {title}")
                created += 1
                # the same CVE can surface twice (two lockfiles, two paths)
                existing[title] = [result]

    # Close issues for CVEs that are no longer present, and duplicates of
    # ones that are (keep the oldest open).
    active_cves = {
        f"[pkg-vuln] {v.get('VulnerabilityID', '')} in "
        f"{v.get('PkgName', '')}@{v.get('InstalledVersion', '')}"
        for v in vulns
    }
    for title, issues in existing.items():
        issues = sorted(issues, key=lambda i: i.get("number", 0))
        _close_all(
            title,
            issues[1:] if title in active_cves else issues,
            dry_run=dry_run,
        )

    return created


def main() -> None:
    import argparse

    parser = argparse.ArgumentParser(
        description="Auto-create Gitea issues for package drift and vulnerabilities"
    )
    parser.add_argument("--drift-only", action="store_true")
    parser.add_argument("--vuln-only", action="store_true")
    parser.add_argument("--dry-run", action="store_true")
    args = parser.parse_args()

    do_drift = not args.vuln_only
    do_vuln = not args.drift_only

    total = 0
    if do_drift:
        print("Checking package drift...")
        total += handle_drift(dry_run=args.dry_run)
    if do_vuln:
        print("Checking vulnerabilities...")
        total += handle_vulns(dry_run=args.dry_run)

    if total:
        print(f"\nCreated {total} issue(s).")
    else:
        print("\nNo new issues to create.")


if __name__ == "__main__":
    main()