"""
Audit the Python environment a build is about to ship against known
vulnerabilities, with pip-audit.

    python audit_deps.py [--tool-dir DIR]

Run with the interpreter of the environment being shipped: what it lists is
what is installed beside it. The builds resolve the newest releases when they
run (there is no lock file), so this is the one moment the shipped set is known.

- A known vulnerability stops the build (exit 1), unless its id is listed in
  packaging/pip-audit-ignore.txt with the reason it does not apply.
- No network is not a reason to stop: the audit is skipped with a warning and
  the build goes on (exit 0). So is a pip-audit that cannot be installed or
  produces no report, which is what a network failing halfway looks like.

pip-audit runs from a venv of its own (--tool-dir, kept between builds), never
from the shipped environment: nothing it pulls in may end up in a package.
Our own packages are left out — they are not on PyPI — and so are pip,
setuptools and wheel, which install the environment and do not ship in it:
build-common.sh strips them from the venv and PyInstaller does not freeze them.
"""

import argparse
import json
import os
import subprocess
import sys
import tempfile
import urllib.request
from importlib import metadata
from pathlib import Path

PIP_AUDIT = "pip-audit==2.10.1"
# What pip-audit asks by default; reachable means the audit can run.
PROBE = "https://pypi.org/pypi/pip-audit/json"
IGNORE_FILE = Path(__file__).resolve().parent / "pip-audit-ignore.txt"
NOT_SHIPPED = {"pip", "setuptools", "wheel"}


def warn(msg: str) -> None:
    print(f"!! pip-audit skipped: {msg}", file=sys.stderr)


def shipped() -> list[str]:
    seen: dict[str, str] = {}
    for dist in metadata.distributions():
        name = (dist.metadata["Name"] or "").strip()
        if name and not name.lower().startswith("meshbay") and name.lower() not in NOT_SHIPPED:
            seen.setdefault(name.lower(), f"{name}=={dist.version}")
    return sorted(seen.values())


def ignored() -> list[str]:
    if not IGNORE_FILE.exists():
        return []
    ids = []
    for line in IGNORE_FILE.read_text(encoding="utf-8").splitlines():
        line = line.split("#", 1)[0].strip()
        if line:
            ids.append(line.split()[0])
    return ids


def tool(tool_dir: Path) -> Path | None:
    exe = tool_dir / ("Scripts/pip-audit.exe" if os.name == "nt" else "bin/pip-audit")
    py = tool_dir / ("Scripts/python.exe" if os.name == "nt" else "bin/python")
    if not py.exists():
        if subprocess.run([sys.executable, "-m", "venv", "--clear", str(tool_dir)],
                          capture_output=True).returncode != 0:
            return None
    r = subprocess.run([str(py), "-m", "pip", "install", "--quiet",
                        "--disable-pip-version-check", PIP_AUDIT],
                       capture_output=True, text=True)
    return exe if r.returncode == 0 and exe.exists() else None


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    ap.add_argument("--tool-dir", type=Path,
                    default=Path(tempfile.gettempdir()) / "meshbay-pip-audit")
    args = ap.parse_args()

    try:
        urllib.request.urlopen(PROBE, timeout=10).close()
    except Exception as e:
        warn(f"no network ({e.__class__.__name__}: {e})")
        return 0

    exe = tool(args.tool_dir)
    if exe is None:
        warn(f"could not install {PIP_AUDIT}")
        return 0

    with tempfile.TemporaryDirectory() as tmp:
        req = Path(tmp) / "shipped.txt"
        req.write_text("\n".join(shipped()) + "\n", encoding="utf-8")
        report = Path(tmp) / "report.json"
        cmd = [str(exe), "-r", str(req), "--no-deps", "--disable-pip",
               "--progress-spinner", "off", "-f", "json", "-o", str(report)]
        for vid in ignored():
            cmd += ["--ignore-vuln", vid]
        run = subprocess.run(cmd, capture_output=True, text=True)
        try:
            data = json.loads(report.read_text(encoding="utf-8"))
        except (OSError, ValueError):
            tail = (run.stderr or run.stdout).strip().splitlines()[-3:]
            warn("no report — " + " | ".join(tail))
            return 0

    # The same id can come back twice, from two advisory sources.
    found: dict[tuple[str, str, str], str] = {}
    for d in data.get("dependencies", []):
        for v in d.get("vulns", []):
            found.setdefault((d["name"], d["version"], v["id"]),
                             ", ".join(v.get("fix_versions") or []) or "none")
    audited = sum(1 for d in data.get("dependencies", []) if "skip_reason" not in d)
    if not found:
        print(f"    pip-audit: {audited} packages, no known vulnerability")
        return 0
    print(f"!! pip-audit: {len(found)} known vulnerabilit{'y' if len(found) == 1 else 'ies'}:",
          file=sys.stderr)
    for (name, version, vid), fix in found.items():
        print(f"!!   {name} {version}  {vid}  (fixed in: {fix})", file=sys.stderr)
    print(f"!! Upgrade, or list the id in {IGNORE_FILE.name} with why it does not apply.",
          file=sys.stderr)
    return 1


if __name__ == "__main__":
    sys.exit(main())