ss-tools/scripts/check_llm_certs.py

#!/usr/bin/env python3
"""
SSL Certificate Diagnostics for superset-tools.

Tests all 4 layers with the actual libraries used in production.
Key finding from ADR-0009: use capath, NOT cafile — OpenSSL 3.x
ignores intermediate CA certs in cafile (verify code 20) but
correctly builds chains with capath (verify code 0).

Output: human-readable + JSON summary with AI diagnosis hint.
"""

from __future__ import annotations

import argparse
import json
import os
import platform
import subprocess
import sys
from pathlib import Path

# ── Constants ──
CA_BUNDLE = Path("/etc/ssl/certs/ca-certificates.crt")
CA_DIR = Path("/etc/ssl/certs")
CUSTOM_DIRS = [
    Path("/usr/local/share/ca-certificates/custom"),
    Path("/usr/local/share/ca-certificates/llm"),
]
NSS_DB = f"sql:{Path(os.environ.get('HOME', '/root')) / '.pki' / 'nssdb'}"

results: dict = {}


# ── Helpers ──


def _run(
    cmd: list[str], timeout: int = 15, input_data: str = ""
) -> tuple[int, str, str]:
    """Run subprocess with optional stdin."""
    try:
        r = subprocess.run(
            cmd, input=input_data, capture_output=True, text=True, timeout=timeout
        )
        return r.returncode, r.stdout, r.stderr
    except FileNotFoundError:
        return -1, "", "command not found"
    except subprocess.TimeoutExpired:
        return -2, "", f"timeout ({timeout}s)"


def _count_certs(path: str) -> int:
    try:
        with open(path, "rb") as f:
            return f.read().count(b"-----BEGIN CERTIFICATE-----")
    except Exception:
        return 0


# ── Openssl tests ──


def test_openssl(label: str, extra_args: list[str]) -> None:
    """Test openssl s_client with empty stdin to prevent hang."""
    cmd = [
        "openssl",
        "s_client",
        "-connect",
        f"{host}:443",
        "-servername",
        host,
        "-verify_return_error",
    ] + extra_args
    rc, out, err = _run(cmd, timeout=20, input_data="")
    for line in out.split("\n"):
        if "Verify return code" in line:
            ok = "0 (ok)" in line
            results[f"openssl_{label}"] = f"{'OK' if ok else 'FAIL'}: {line.strip()}"
            print(f"  openssl_{label}: {'✅' if ok else '❌'} {line.strip()}")
            return
    results[f"openssl_{label}"] = f"ERROR: {err.strip() or 'no output'}"
    print(f"  openssl_{label}: ❌ {results[f'openssl_{label}']}")


# ── Python library tests ──


def test_httpx(key: str, verify) -> None:
    try:
        import httpx

        r = httpx.get(TARGET_URL, verify=verify, timeout=10)
        results[key] = "OK"
        print(f"  {key}: ✅ status={r.status_code}")
    except Exception as e:
        results[key] = f"FAIL: {type(e).__name__}"
        print(f"  {key}: ❌ {type(e).__name__}")


def test_requests(key: str, verify) -> None:
    try:
        import requests as req

        r = req.get(TARGET_URL, verify=verify, timeout=10)
        results[key] = "OK"
        print(f"  {key}: ✅ status={r.status_code}")
    except Exception as e:
        results[key] = f"FAIL: {type(e).__name__}"
        print(f"  {key}: ❌ {type(e).__name__}")


# ══════════════════════════════════════════════
#  MAIN
# ══════════════════════════════════════════════

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--target", default=os.environ.get("TARGET_URL", "https://lite.ai.rusal.com")
    )
    args = parser.parse_args()
    TARGET_URL = args.target
    host = TARGET_URL.replace("https://", "").split("/")[0]

    # ── 1. System ──
    print("\n=== 1. System ===")
    print(f"  target={TARGET_URL}  host={host}")
    print(f"  platform={platform.platform()}")
    print(f"  python={sys.version.split()[0]}")
    for lib in ("httpx", "requests", "openai", "certifi"):
        try:
            mod = __import__(lib)
            v = getattr(mod, "__version__", "installed")
            p = getattr(mod, "__file__", "")
            if lib == "certifi":
                print(
                    f"  {lib}={v}  path={mod.where()}  certs={_count_certs(mod.where())}"
                )
            else:
                print(f"  {lib}={v}")
        except ImportError:
            print(f"  {lib}=NOT_INSTALLED")

    # ── 2. System CA Store ──
    print("\n=== 2. System CA Store ===")
    bundle_certs = _count_certs(str(CA_BUNDLE)) if CA_BUNDLE.exists() else 0
    hash_links = len(list(CA_DIR.glob("[0-9a-f]*.[0-9]"))) if CA_DIR.is_dir() else 0
    print(f"  bundle={CA_BUNDLE}  certs={bundle_certs}")
    print(f"  capath={CA_DIR}  hash_links={hash_links}")

    print("  custom certs:")
    for d in CUSTOM_DIRS:
        if not d.is_dir():
            continue
        for f in sorted(d.iterdir()):
            if f.suffix in (".pem", ".crt") and f.is_file():
                rc, out, _ = _run(
                    ["openssl", "x509", "-in", str(f), "-noout", "-subject", "-issuer"],
                    timeout=3,
                )
                if rc != 0:
                    rc, out, _ = _run(
                        [
                            "openssl",
                            "x509",
                            "-in",
                            str(f),
                            "-inform",
                            "DER",
                            "-noout",
                            "-subject",
                        ],
                        timeout=3,
                    )
                subj = out.replace("subject=", "").strip() if rc == 0 else "UNREADABLE"
                print(f"    {f.parent.name}/{f.name}  subj={subj[:60]}")

    # ── 3. OpenSSL s_client (3 variants) ──
    print("\n=== 3. OpenSSL s_client ===")
    test_openssl("default", [])
    test_openssl("cafile", ["-CAfile", str(CA_BUNDLE)])
    test_openssl("capath", ["-CApath", str(CA_DIR)])

    # ── 4. Python httpx ──
    print("\n=== 4. Python httpx (LLMClient) ===")
    try:
        import httpx

        test_httpx("httpx_True", verify=True)
        try:
            import ssl

            ctx_cafile = (
                ssl.create_default_context(cafile=str(CA_BUNDLE))
                if CA_BUNDLE.exists()
                else ssl.create_default_context()
            )
            ctx_capath = (
                ssl.create_default_context(capath=str(CA_DIR))
                if CA_DIR.is_dir()
                else ssl.create_default_context()
            )
            test_httpx("httpx_cafile", verify=ctx_cafile)
            test_httpx("httpx_capath", verify=ctx_capath)
        except Exception as e:
            print(f"  SSLContext: ❌ {e}")
        test_httpx("httpx_False", verify=False)
    except ImportError:
        print("  httpx: NOT_INSTALLED")

    # ── 5. Python requests ──
    print("\n=== 5. Python requests (translate plugin) ===")
    try:
        import requests as req

        test_requests("requests_True", verify=True)
        test_requests("requests_cafile", verify=str(CA_BUNDLE))
        test_requests("requests_capath", verify=str(CA_DIR))
        test_requests("requests_False", verify=False)
    except ImportError:
        print("  requests: NOT_INSTALLED")

    # ── 6. NSS Chromium ──
    print("\n=== 6. NSS Chromium ===")
    rc, out, err = _run(["certutil", "-L", "-d", NSS_DB], timeout=5)
    if rc == 0:
        lines = [
            l for l in out.split("\n") if l.strip() and "Certificate Nickname" not in l
        ]
        print(f"  DB={NSS_DB}  certs={len(lines)}")
        for l in lines:
            if any(x in l.lower() for x in ("rusal", "llm-", "custom-")):
                print(f"    {l}")
    else:
        print(f"  NSS: {err.strip() or 'unavailable'}")

    # ── 7. Centralized SSL (core.ssl) ──
    print("\n=== 7. Centralized SSL (core.ssl) ===")
    try:
        from src.core.ssl import system_ssl_context, describe_context

        ctx = system_ssl_context()
        print(f"  {describe_context(ctx)}")
        print(
            "  LLM_SSL_VERIFY and LLM_CA_CERT_URLS removed — centralized CERTS_PATH only."
        )
    except ImportError:
        print("  src.core.ssl: UNAVAILABLE (backend not installed)")

    # ── 8. JSON Summary ──
    print("\n=== 8. JSON Summary ===")
    summary = {
        k: results[k]
        for k in [
            "openssl_default",
            "openssl_cafile",
            "openssl_capath",
            "httpx_True",
            "httpx_cafile",
            "httpx_capath",
            "httpx_False",
            "requests_True",
            "requests_cafile",
            "requests_capath",
            "requests_False",
        ]
        if k in results
    }

    # Diagnosis
    diag = []
    for k in ("openssl_default", "openssl_cafile", "openssl_capath"):
        v = str(summary.get(k, ""))
        diag.append(f"{k}={'OK' if v.startswith('OK') else 'FAIL'}")
    for k in sorted(summary):
        if k.startswith("httpx") or k.startswith("requests"):
            v = str(summary.get(k, ""))
            diag.append(f"{k}={'OK' if v.startswith('OK') else 'FAIL'}")

    if str(results.get("httpx_capath", "")).startswith("OK") and not str(
        results.get("httpx_cafile", "")
    ).startswith("OK"):
        diag.append("VERDICT=capath_works_cafile_fails_use_capath")
    elif str(summary.get("httpx_False", "")).startswith("OK"):
        diag.append("VERDICT=all_ssl_fails_verify_disabled_works")
    elif str(summary.get("openssl_default", "")).startswith("OK"):
        diag.append("VERDICT=all_ok")
    else:
        diag.append("VERDICT=needs_investigation")

    summary["diagnosis"] = "; ".join(diag)
    print(json.dumps(summary, indent=2))