#!/usr/bin/env python3
"""Find every CA trust store on a machine (or in an unpacked container image)
and report the ones that still trust certificate authorities that should no
longer be trusted.

Read-only and offline: the script only reads files and runs `uv --version`.
It never connects to the network and never changes a trust store.

Zero Hunt Research, free self-check (MIT licence):
https://zerohunt.ai/self-check/

Run it with -S, so Python does not process .pth files at start-up (a host
hit by the LiteLLM 1.107.0 compromise runs the stealer on every normal start):

Usage:
    python3 -S trust_store_audit.py              # this machine's Python + OS stores
    python3 -S trust_store_audit.py --scan /srv     # also search a directory tree
    python3 -S trust_store_audit.py --scan ./rootfs # an exported container image
    python3 -S trust_store_audit.py --json          # machine-readable output

Exit codes: 0 nothing found, 1 something to fix, 2 usage error.
"""
from __future__ import annotations

import argparse
import base64
import hashlib
import json
import os
import re
import shutil
import site
import ssl
import subprocess
import sys
from dataclasses import dataclass, field, asdict
from pathlib import Path

__version__ = "0.1.0"

# SHA-256 fingerprints of whole DER certificates. Each entry says why it is
# listed and what to do; "critical" means a fixed release removes it, "review"
# means the right action depends on the trust store (see the note).
DISTRUSTED = {
    "86A1ECBA089C4A8D3BBE2734C612BA341D813E043CF9E8A862CD5C57A36BBE6B": {
        "name": "emSign ECC Root CA - G3",
        "severity": "critical",
        "incident": "eMudhra 2026-10: eight intermediate CA keys recoverable from their public keys",
        "advice": "Update the bundle that contains it (certifi >= 2026.10.19, pip >= 26.2.2, "
                  "the OS ca-certificates package) and rebuild images that vendor it.",
        "reference": "https://github.com/advisories/GHSA-6q48-6c7w-9gw2",
    },
    "D7A7A0FB5D7E2731D771E9484EBCDEF71D5F0C3E0A2948782BC83EE0EA699EF4": {
        "name": "AAA Certificate Services (Sectigo)",
        "severity": "review",
        "incident": "eMudhra 2026-10: six of the eight compromised intermediates are cross-signed from this root",
        "advice": "Check that your OS vendor's updated ca-certificates removes or blocks the cross-sign path; "
                  "if no update exists yet, apply the vendor's blocklist instructions.",
        "reference": "https://emudhra.we-have-a-problem.com/",
    },
}

# First release that no longer trusts the eMudhra chain.
FIXED_VERSIONS = {
    "certifi": (2026, 10, 19),
    "pip": (26, 2, 2),
    "uv": (0, 13, 4),
}

OS_BUNDLES = [
    "/etc/ssl/certs/ca-certificates.crt",      # Debian, Ubuntu, Alpine
    "/etc/pki/tls/certs/ca-bundle.crt",        # RHEL, Fedora, Amazon Linux
    "/etc/pki/ca-trust/extracted/pem/tls-ca-bundle.pem",
    "/etc/ssl/ca-bundle.pem",                  # openSUSE
    "/etc/ssl/cert.pem",                       # macOS, Alpine, BSD
    "/usr/local/etc/openssl/cert.pem",         # Homebrew OpenSSL
    "/opt/homebrew/etc/ca-certificates/cert.pem",
]

ENV_VARS = ["SSL_CERT_FILE", "REQUESTS_CA_BUNDLE", "CURL_CA_BUNDLE", "PIP_CERT",
            "NODE_EXTRA_CA_CERTS", "GIT_SSL_CAINFO", "AWS_CA_BUNDLE"]

BUNDLE_NAME = re.compile(r"(cacert.*\.pem|ca-bundle.*\.(pem|crt)|ca-certificates\.crt|cert\.pem|"
                         r"tls-ca-bundle\.pem|.*roots?\.pem)$", re.I)
PEM_BLOCK = re.compile(rb"-----BEGIN CERTIFICATE-----\s+(.+?)\s+-----END CERTIFICATE-----", re.S)
MAX_BUNDLE_BYTES = 8 * 1024 * 1024
SKIP_DIRS = {"proc", "sys", "dev", "run", ".git", "node_modules"}


@dataclass
class Finding:
    path: str
    kind: str
    certificates: int = 0
    distrusted: list = field(default_factory=list)
    version: str | None = None
    outdated: str | None = None

    @property
    def bad(self) -> bool:
        return bool(self.distrusted or self.outdated)


def fingerprints(data: bytes) -> list[str]:
    out = []
    for block in PEM_BLOCK.findall(data):
        try:
            der = base64.b64decode(b"".join(block.split()), validate=True)
        except ValueError:
            continue
        out.append(hashlib.sha256(der).hexdigest().upper())
    return out


def parse_version(text: str) -> tuple[int, ...] | None:
    m = re.search(r"(\d+(?:\.\d+)+)", text)
    return tuple(int(x) for x in m.group(1).split(".")) if m else None


def outdated(tool: str, version: str | None) -> str | None:
    v = parse_version(version or "")
    fixed = FIXED_VERSIONS[tool]
    if v is None or v >= fixed:
        return None
    return f"{tool} {version} < {'.'.join(map(str, fixed))}"


def read_version(init_py: Path) -> str | None:
    try:
        m = re.search(r"__version__\s*=\s*['\"]([^'\"]+)", init_py.read_text(errors="replace"))
    except OSError:
        return None
    return m.group(1) if m else None


def audit_bundle(path: Path, kind: str) -> Finding | None:
    try:
        if not path.is_file() or path.stat().st_size > MAX_BUNDLE_BYTES:
            return None
        data = path.read_bytes()
    except OSError:
        return None
    fps = fingerprints(data)
    if not fps:
        return None
    f = Finding(path=str(path), kind=kind, certificates=len(fps))
    f.distrusted = [dict(fingerprint=fp, **DISTRUSTED[fp]) for fp in fps if fp in DISTRUSTED]
    # certifi and pip's vendored copy carry a version next to the bundle.
    pkg = path.parent
    if pkg.name == "certifi":
        f.version = read_version(pkg / "__init__.py")
        f.outdated = outdated("certifi", f.version)
        if pkg.parent.name == "_vendor" and pkg.parent.parent.name == "pip":
            f.kind = "pip (vendored certifi)"
            pip_version = read_version(pkg.parent.parent / "__init__.py")
            f.version = f"pip {pip_version}, certifi {f.version}"
            f.outdated = outdated("pip", pip_version) or f.outdated
    return f


def python_bundles() -> list[tuple[Path, str]]:
    roots = set(sys.path) | set(site.getsitepackages() if hasattr(site, "getsitepackages") else [])
    roots.add(site.getusersitepackages())
    found = []
    for root in roots:
        base = Path(root)
        for rel, kind in (("certifi/cacert.pem", "certifi"),
                          ("pip/_vendor/certifi/cacert.pem", "pip (vendored certifi)")):
            if (base / rel).is_file():
                found.append((base / rel, kind))
    return found


def system_bundles() -> list[tuple[Path, str]]:
    found = [(Path(p), "OS trust store") for p in OS_BUNDLES]
    paths = ssl.get_default_verify_paths()
    for p in (paths.cafile, paths.openssl_cafile):
        if p:
            found.append((Path(p), "OpenSSL default"))
    for var in ENV_VARS:
        if os.environ.get(var):
            found.append((Path(os.environ[var]), f"${var}"))
    return found


def scan_tree(root: Path) -> list[tuple[Path, str]]:
    found = []
    for dirpath, dirnames, filenames in os.walk(root, followlinks=False):
        dirnames[:] = [d for d in dirnames if d not in SKIP_DIRS]
        for name in filenames:
            if BUNDLE_NAME.match(name):
                found.append((Path(dirpath) / name, "found by --scan"))
    return found


def uv_check() -> Finding | None:
    exe = shutil.which("uv")
    if not exe:
        return None
    try:
        out = subprocess.run([exe, "--version"], capture_output=True, text=True, timeout=10).stdout
    except (OSError, subprocess.SubprocessError):
        return None
    version = out.split()[1] if len(out.split()) > 1 else out.strip()
    return Finding(path=exe, kind="uv (bundled webpki roots)", version=version,
                   outdated=outdated("uv", version))


def audit(scan: list[Path], include_local: bool = True) -> list[Finding]:
    candidates: list[tuple[Path, str]] = []
    if include_local:
        candidates += python_bundles() + system_bundles()
    for root in scan:
        candidates += scan_tree(root)
    seen, results = set(), []
    for path, kind in candidates:
        try:
            key = path.resolve()
        except OSError:
            continue
        if key in seen:
            continue
        seen.add(key)
        f = audit_bundle(path, kind)
        if f:
            results.append(f)
    if include_local:
        uv = uv_check()
        if uv:
            results.append(uv)
    return results


def print_report(results: list[Finding]) -> None:
    if not results:
        print("No CA bundles found.")
        return
    for f in results:
        status = "FIX " if f.bad else "ok  "
        detail = f"{f.certificates} certs" if f.certificates else ""
        if f.version:
            detail += (", " if detail else "") + f.version
        print(f"[{status}] {f.kind}: {f.path} ({detail})")
        if f.outdated:
            print(f"        outdated: {f.outdated}")
        for d in f.distrusted:
            print(f"        {d['severity'].upper()}: trusts {d['name']}")
            print(f"          {d['incident']}")
            print(f"          {d['advice']}")
            print(f"          {d['reference']}")
    bad = sum(f.bad for f in results)
    print(f"\n{len(results)} trust stores checked, {bad} to fix.")
    if bad:
        print("Remediation guide: https://zerohunt.ai/blog/"
              "emudhra-ca-keys-python-certifi-pip-trust-store-remediation/#remediation")


def main(argv: list[str] | None = None) -> int:
    ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
    ap.add_argument("--scan", action="append", default=[], type=Path, metavar="DIR",
                    help="also search this directory tree for CA bundles (repeatable)")
    ap.add_argument("--only-scan", action="store_true",
                    help="skip this machine's Python and OS stores; check only --scan trees")
    ap.add_argument("--json", action="store_true", help="print JSON instead of text")
    ap.add_argument("--version", action="version", version=f"%(prog)s {__version__}")
    args = ap.parse_args(argv)
    if args.only_scan and not args.scan:
        ap.error("--only-scan needs at least one --scan DIR")
    for d in args.scan:
        if not d.is_dir():
            ap.error(f"not a directory: {d}")
    results = audit(args.scan, include_local=not args.only_scan)
    if args.json:
        print(json.dumps([dict(asdict(f), needs_fix=f.bad) for f in results], indent=2))
    else:
        print_report(results)
    return 1 if any(f.bad for f in results) else 0


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