ss-tools/scripts/diag_container.py

#!/usr/bin/env python3
"""Diagnostic script: verify SSL cert installation and encryption key health.

Usage:
    docker cp scripts/diag_container.py superset-tools-backend-1:/tmp/
    docker compose exec backend python3 /tmp/diag_container.py [--target https://lite.ai.rusal.com]
"""

import hashlib
import os
import ssl
import subprocess
import sys
import traceback
from pathlib import Path

# ── Console helpers ──────────────────────────────────────────────────
GREEN = "\033[32m"
RED = "\033[31m"
YELLOW = "\033[33m"
BOLD = "\033[1m"
RESET = "\033[0m"


def ok(msg: str) -> None:
    print(f"  {GREEN}✅{RESET} {msg}")


def fail(msg: str) -> None:
    print(f"  {RED}❌{RESET} {msg}")


def warn(msg: str) -> None:
    print(f"  {YELLOW}⚠️ {RESET} {msg}")


def section(title: str) -> None:
    print(f"\n{BOLD}{'=' * 60}{RESET}")
    print(f"{BOLD}  {title}{RESET}")
    print(f"{BOLD}{'=' * 60}{RESET}")


# ── Section 1: Environment ───────────────────────────────────────────


def check_env() -> None:
    section("1. Environment variables")

    for var in (
        "ENCRYPTION_KEY",
        "AUTH_SECRET_KEY",
        "LLM_SSL_VERIFY",
        "LLM_CA_CERT_URLS",
    ):
        val = os.getenv(var, "")
        if not val:
            warn(f"{var} is NOT set")
        elif var in ("ENCRYPTION_KEY", "AUTH_SECRET_KEY"):
            fp = hashlib.sha256(val.encode()).hexdigest()[:8]
            ok(f"{var} is set (sha256:{fp})")
        else:
            ok(f"{var} = {val}")

    fp = hashlib.sha256(os.getenv("ENCRYPTION_KEY", "").encode()).hexdigest()[:8]
    print(f"\n  Current key fingerprint: {BOLD}sha256:{fp}{RESET}")

    if os.getenv("LLM_SSL_VERIFY", "").lower() in ("false", "0", "no", "off"):
        fail(
            "LLM_SSL_VERIFY is DISABLED — SSL verification bypassed.\n"
            "  This masks the root cause. Remove LLM_SSL_VERIFY=false\n"
            "  and fix cert installation instead."
        )
    else:
        ok(
            "LLM_SSL_VERIFY is ENABLED (capath approach).\n"
            "  If LLM calls fail, the certificate for the target server\n"
            "  is not in /etc/ssl/certs/."
        )


# ── Section 2: System CA store ──────────────────────────────────────


def check_system_certs() -> None:
    section("2. System CA store (/etc/ssl/certs/)")

    bundle = Path("/etc/ssl/certs/ca-certificates.crt")
    if not bundle.exists():
        fail("ca-certificates.crt NOT found — cert installation failed fatally")
        return

    cert_count = subprocess.run(
        ["grep", "-c", "BEGIN CERTIFICATE", str(bundle)],
        capture_output=True,
        text=True,
    ).stdout.strip()
    ok(f"ca-certificates.crt exists, {cert_count} certificates in bundle")

    certs_dir = Path("/etc/ssl/certs")
    custom_pems = list(Path("/usr/local/share/ca-certificates").rglob("*.crt"))
    llm_pems = list(Path("/usr/local/share/ca-certificates/llm").rglob("*.crt"))

    if custom_pems:
        names = [p.name for p in custom_pems]
        ok(f"Installed CA certs: {', '.join(names)}")
    else:
        warn("No custom .crt files in /usr/local/share/ca-certificates/")

    if llm_pems:
        names = [p.name for p in llm_pems]
        ok(f"LLM CA certs: {', '.join(names)}")
    else:
        warn(
            "No LLM CA certs downloaded.\n"
            "  LLM_CA_CERT_URLS is not set — set it in .env.enterprise-clean\n"
            "  or place the CA .crt in ./certs/ on the host."
        )

    # Check hash symlinks
    hash_links = [p for p in certs_dir.iterdir() if p.is_symlink()]
    llm_links = [l for l in hash_links if "llm" in str(Path(os.readlink(str(l))))]
    if llm_links:
        ok(f"Hash symlinks for LLM certs: {len(llm_links)}")
    else:
        warn("No hash symlinks pointing to LLM cert dir — capath may not find them")


# ── Section 3: OpenSSL connectivity ──────────────────────────────────


def check_openssl(target: str) -> None:
    section(f"3. OpenSSL connectivity ({target})")

    # cafile approach (expected to FAIL with code 20 per ADR-0009 Finding 7)
    cafile = "/etc/ssl/certs/ca-certificates.crt"
    r = subprocess.run(
        [
            "openssl",
            "s_client",
            "-connect",
            target,
            "-servername",
            target.split(":")[0],
            "-CAfile",
            cafile,
        ],
        input="",
        capture_output=True,
        text=True,
        timeout=15,
    )
    if "Verify return code: 0" in r.stderr or "Verify return code: 0" in r.stdout:
        ok(f"cafile: verify OK (unexpected — usually fails per ADR-0009 Finding 7)")
    elif "Verify return code: 20" in (r.stderr + r.stdout):
        warn(
            f"cafile: verify code 20 (EXPECTED per ADR-0009 — intermediate CA ignored)"
        )
    else:
        fail(f"cafile: unexpected output, returncode={r.returncode}")

    # capath approach (should succeed)
    capath = "/etc/ssl/certs/"
    r = subprocess.run(
        [
            "openssl",
            "s_client",
            "-connect",
            target,
            "-servername",
            target.split(":")[0],
            "-CApath",
            capath,
        ],
        input="",
        capture_output=True,
        text=True,
        timeout=15,
    )
    if "Verify return code: 0" in (r.stderr + r.stdout):
        ok(f"capath: verify OK — cert chain trusted")
    else:
        fail(
            f"capath: verify FAILED.\n"
            f"  The CA certificate for {target} is NOT in /etc/ssl/certs/.\n"
            f"  Add it via LLM_CA_CERT_URLS or ./certs/ on the host."
        )


# ── Section 4: Python SSL context ────────────────────────────────────


def check_python_ssl(target: str) -> None:
    section(f"4. Python SSL context ({target})")

    # capath approach (what _get_ssl_verify() uses)
    ctx = ssl.create_default_context(capath="/etc/ssl/certs/")
    host, port = target.split(":") if ":" in target else (target, 443)
    port = int(port)

    try:
        with ctx.wrap_socket(__import__("socket").socket(), server_hostname=host) as s:
            s.settimeout(10)
            s.connect((host, port))
            cert = s.getpeercert()
            cn = dict(x[0] for x in cert.get("subject", []))
            ok(f"SSLContext(capath): connected OK — CN={cn.get('commonName', '?')}")
    except Exception as e:
        fail(f"SSLContext(capath): FAILED — {e}")


# ── Section 5: httpx connectivity ────────────────────────────────────


def check_httpx(target: str) -> None:
    section(f"5. httpx connectivity ({target})")

    try:
        import httpx

        ctx = ssl.create_default_context(capath="/etc/ssl/certs/")
        url = f"https://{target}" if not target.startswith("http") else target
        r = httpx.get(url, verify=ctx, timeout=10, follow_redirects=True)
        ok(f"httpx(capath): HTTP {r.status_code}")
    except ImportError:
        warn("httpx not installed — skipping")
    except Exception as e:
        fail(f"httpx(capath): FAILED — {e}")


# ── Section 6: Encryption health ─────────────────────────────────────


def check_encryption_health() -> None:
    section("6. Encryption key health (internal)")

    sys.path.insert(0, "/app/backend")
    try:
        from src.core.encryption import get_encryption_manager, is_fernet_token
        from src.core.database import SessionLocal
        from src.models.llm import LLMProvider

        mgr = get_encryption_manager()
        db = SessionLocal()
        try:
            providers = db.query(LLMProvider).all()
            ok(f"LLM providers found: {len(providers)}")
            for p in providers:
                if not p.api_key:
                    warn(f"  {p.name}: api_key is EMPTY — no key stored")
                    continue
                if not is_fernet_token(p.api_key):
                    warn(
                        f"  {p.name}: api_key is NOT a fernet token — stored unencrypted?"
                    )
                    continue
                try:
                    mgr.decrypt(p.api_key)
                    ok(f"  {p.name}: api_key decrypts OK (healthy)")
                except Exception as e:
                    fail(
                        f"  {p.name}: api_key decrypt FAILED — "
                        f"encrypted with different ENCRYPTION_KEY"
                    )
        finally:
            db.close()
    except ImportError as e:
        warn(f"Cannot import backend modules: {e}")
    except Exception as e:
        fail(f"Encryption health check failed: {e}")


# ── Main ─────────────────────────────────────────────────────────────


def main() -> None:
    target = "lite.ai.rusal.com:443"
    for i, arg in enumerate(sys.argv[1:]):
        if arg == "--target" and i + 1 < len(sys.argv):
            target = sys.argv[i + 1]
        elif arg.startswith("--"):
            print(f"Unknown flag: {arg}")
            print(f"Usage: {sys.argv[0]} [--target host:port]")
            sys.exit(1)

    print(f"\n{BOLD}superset-tools container diagnostic{RESET}")
    print(f"Target: {target}\n")

    check_env()
    check_system_certs()
    check_openssl(target)
    check_python_ssl(target)
    check_httpx(target)
    check_encryption_health()

    section("Summary")
    print(
        "\nExpected (per ADR-0009):\n"
        "  - openssl CAfile:   code 20 FAIL (intermediate CA ignored)\n"
        "  - openssl CApath:   code 0  OK\n"
        "  - Python SSLContext(capath): OK\n"
        "  - httpx(capath):    HTTP 200 OK\n"
        "\nIf CApath/httpx failures:\n"
        "  → LLM_CA_CERT_URLS not set → certs not downloaded\n"
        "  → set LLM_CA_CERT_URLS or add .crt to ./certs/\n"
    )

    print(
        "\nIf ENCRYPTION_KEY health failures:\n"
        "  → Stored secrets encrypted with old key\n"
        "  → Open Settings → System → 'Encryption Key Recovery'\n"
        "  → Re-enter LLM API keys and DB passwords\n"
    )


if __name__ == "__main__":
    try:
        main()
    except Exception:
        print(f"\n{RED}Script crashed:{RESET}")
        traceback.print_exc()
        sys.exit(1)