diff --git a/scripts/check_llm_certs.py b/scripts/check_llm_certs.py index 74acaafb..011e28dd 100755 --- a/scripts/check_llm_certs.py +++ b/scripts/check_llm_certs.py @@ -1,25 +1,13 @@ #!/usr/bin/env python3 # #region check_llm_certs [C:4] [TYPE Module] [SEMANTICS ssl, certs, diagnostics, python] -# @BRIEF Diagnóstico SSL pro ss-tools em Python. Testa todas as bibliotecas -# reais usadas no projeto: ssl, httpx, requests, openai, certifi. -# Saída em formato parseável para AI corrigir o código. -# @USAGE ./scripts/check_llm_certs.py [--target https://...] [-v] +# @BRIEF SSL diagnostics in Python — tests all real libraries used by ss-tools: +# ssl, httpx (as LLMClient), requests (as translate), openai, certifi, NSS. +# Output is AI-parseable to fix the code. +# @USAGE ./scripts/check_llm_certs.py [--target https://...] # @LAYER Infrastructure # #endregion check_llm_certs -""" -SSL Certificate Diagnostic Tool for ss-tools. - -Tests all HTTP stacks used by the application: - - ssl (Python standard library) - - httpx (used by LLMClient in llm_analysis plugin) - - openai.AsyncOpenAI (actual LLM client wrapper) - - requests (used by translate plugin) - - certifi (default CA bundle used by requests/httpx) - - NSS certutil (Chromium/Playwright) - -Output format: key=value lines + JSON summary for AI processing. -""" +from __future__ import annotations import json import os @@ -27,9 +15,10 @@ import platform import subprocess import sys import tempfile +import traceback from pathlib import Path -# ── Configuration ── +# ── Configuration (overridable via env) ── TARGET_URL = os.environ.get("TARGET_URL", "https://lite.ai.rusal.com") LLM_SSL_VERIFY = os.environ.get("LLM_SSL_VERIFY", "true") CA_BUNDLE = Path("/etc/ssl/certs/ca-certificates.crt") @@ -41,266 +30,35 @@ CUSTOM_DIRS = [ NSS_DB_DIR = Path(os.environ.get("HOME", "/root")) / ".pki" / "nssdb" results: dict = {} -errors: list = [] -def section(name: str) -> None: - """Print section header.""" - print(f"\n=== {name} ===") +# ── Lazy library loader ── +_LIBS: dict = {} -def fmt_ok(val: str = "OK") -> str: - return f"✅ {val}" - - -def fmt_fail(val: str) -> str: - return f"❌ {val}" - - -def fmt_warn(val: str) -> str: - return f"⚠ {val}" - - -def run_cmd(cmd: list[str], timeout: int = 15) -> tuple[int, str, str]: - """Run shell command, return (returncode, stdout, stderr).""" +def _import(name: str) -> str | None: + """Lazy import, returns error message or None on success.""" + if name in _LIBS: + return _LIBS[name] if isinstance(_LIBS[name], str) else None try: - r = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout) - return r.returncode, r.stdout, r.stderr - except FileNotFoundError: - return -1, "", f"command not found: {cmd[0]}" - except subprocess.TimeoutExpired: - return -2, "", "timeout" + __import__(name) + _LIBS[name] = True + return None + except ImportError: + _LIBS[name] = f"NOT_INSTALLED" + return _LIBS[name] -# ═══════════════════════════════════════════════════════ -# 1. SYSTEM INFORMATION -# ═══════════════════════════════════════════════════════ -section("1. System") -host = TARGET_URL.replace("https://", "").split("/")[0] -results["target_url"] = TARGET_URL -results["host"] = host -results["llm_ssl_verify"] = LLM_SSL_VERIFY -results["platform"] = platform.platform() -results["python_version"] = sys.version +def _lib_ver(name: str) -> str: + err = _import(name) + if err: + return err + mod = __import__(name) + v = getattr(mod, "__version__", None) + return str(v) if v else "installed" -# Libraries -try: - import httpx as _httpx - results["httpx_version"] = _httpx.__version__ -except Exception as e: - results["httpx_version"] = f"NOT_INSTALLED ({e})" - -try: - import requests as _requests - results["requests_version"] = _requests.__version__ -except Exception as e: - results["requests_version"] = f"NOT_INSTALLED ({e})" - -try: - import certifi as _certifi - results["certifi_path"] = _certifi.where() - results["certifi_certs"] = _count_certs(_certifi.where()) -except Exception as e: - results["certifi_path"] = f"NOT_INSTALLED ({e})" - -try: - import openai as _openai - results["openai_version"] = _openai.__version__ -except Exception as e: - results["openai_version"] = f"NOT_INSTALLED ({e})" - - -# ═══════════════════════════════════════════════════════ -# 2. SYSTEM CA STORE -# ═══════════════════════════════════════════════════════ -section("2. System CA Store") - -results["ca_bundle"] = str(CA_BUNDLE) -results["ca_bundle_exists"] = CA_BUNDLE.exists() -results["ca_dir_exists"] = CA_DIR.is_dir() - -if CA_BUNDLE.exists(): - results["ca_bundle_certs"] = _count_certs(str(CA_BUNDLE)) -else: - results["ca_bundle_certs"] = 0 - -if CA_DIR.is_dir(): - hash_links = list(CA_DIR.glob("[0-9a-f]*.[0-9]")) - results["ca_hash_links"] = len(hash_links) -else: - results["ca_hash_links"] = 0 - -print(f" CA bundle: {CA_BUNDLE} ({results['ca_bundle_certs']} certs, exists={results['ca_bundle_exists']})") -print(f" CA dir: {CA_DIR} ({results['ca_hash_links']} hash links)") -print(f" certifi: {results.get('certifi_path', 'N/A')} ({results.get('certifi_certs', 'N/A')} certs)") - -# Custom certs -print(" Custom certs:") -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(): - info = _cert_info(f) - custom_certs.append(info) - print(f" {info['file']}") - print(f" subject={info['subject']}") - print(f" issuer={info['issuer']}") - print(f" fingerprint={info['fingerprint']}") - print(f" in_bundle={info['in_bundle']}") - -results["custom_certs_count"] = len(custom_certs) -results["custom_certs"] = custom_certs - - -# ═══════════════════════════════════════════════════════ -# 3. OPENSSL s_client TESTS -# ═══════════════════════════════════════════════════════ -section("3. OpenSSL s_client") - -# A) Default (system trust store) -rc, out, err = run_cmd([ - "openssl", "s_client", "-connect", f"{host}:443", - "-servername", host, - " 5] - results["nss_certs_count"] = len(nss_certs) if len(nss_certs) > 1 else 0 - print(f" NSS DB: {NSS_DB_DIR} ({results['nss_certs_count']} certs)") - for c in nss_certs: - if any(x in c.lower() for x in ("rusal", "llm-", "custom-")): - print(f" {c}") -else: - results["nss_available"] = False - results["nss_note"] = err.strip() or "certutil not available or NSS DB not found" - print(f" NSS: {results['nss_note']}") - - -# ═══════════════════════════════════════════════════════ -# 7. LLM SSL VERIFY ENV -# ═══════════════════════════════════════════════════════ -section("7. LLM_SSL_VERIFY") - -raw = LLM_SSL_VERIFY.strip().lower() -disabled = raw in ("false", "0", "no", "off") -results["llm_ssl_verify_raw"] = raw -results["llm_ssl_verify_disabled"] = disabled -print(f" LLM_SSL_VERIFY={raw} → {'DISABLED' if disabled else 'ENABLED'}") - -if not disabled: - try: - ctx = _make_ctx_cafile() - results["ssl_context_works"] = True - print(f" SSLContext(cafile=...) → OK (verify_mode={ctx.verify_mode})") - except Exception as e: - results["ssl_context_works"] = False - results["ssl_context_error"] = str(e) - print(f" SSLContext(cafile=...) → FAIL: {e}") - - -# ═══════════════════════════════════════════════════════ -# 8. JSON SUMMARY -# ═══════════════════════════════════════════════════════ -section("8. JSON Summary") - -# Simplify results for AI -summary = { - key: results[key] - for key in [ - "target_url", "host", "llm_ssl_verify", "llm_ssl_verify_disabled", - "ca_bundle_certs", "ca_hash_links", - "custom_certs_count", - "openssl_default", "openssl_cafile", "openssl_capath", - "httpx_true", "httpx_cafile", "httpx_cafile_capath", "httpx_false", - "requests_true", "requests_cafile", "requests_capath", "requests_false", - "nss_available", "nss_certs_count", - "ssl_context_works", - ] - if key in results -} -summary["ai_diagnosis"] = _diagnose(summary) - -print(json.dumps(summary, indent=2, ensure_ascii=False)) - - -# ═══════════════════════════════════════════════════════ -# HELPER FUNCTIONS -# ═══════════════════════════════════════════════════════ def _count_certs(path: str) -> int: - """Count PEM certificates in a file.""" try: with open(path) as f: return f.read().count("-----BEGIN CERTIFICATE-----") @@ -309,78 +67,66 @@ def _count_certs(path: str) -> int: def _cert_info(path: Path) -> dict: - """Extract certificate info.""" import subprocess as sp info = {"file": str(path), "format": path.suffix} - r = sp.run( - ["openssl", "x509", "-in", str(path), "-noout", "-subject", "-issuer", - "-fingerprint", "-sha256"], + ["openssl", "x509", "-in", str(path), "-noout", + "-subject", "-issuer", "-fingerprint", "-sha256"], capture_output=True, text=True, timeout=5, ) if r.returncode == 0: - lines = r.stdout.strip().split("\n") - for line in lines: - if line.startswith("subject="): - info["subject"] = line[8:] - elif line.startswith("issuer="): - info["issuer"] = line[7:] - elif line.startswith("SHA256 Fingerprint="): - info["fingerprint"] = line[19:] + for line in r.stdout.strip().split("\n"): + if line.startswith("subject="): info["subject"] = line[8:] + elif line.startswith("issuer="): info["issuer"] = line[7:] + elif line.startswith("SHA256 Fingerprint="): info["fingerprint"] = line[19:] else: - # Try DER format - r2 = sp.run( - ["openssl", "x509", "-in", str(path), "-inform", "DER", - "-noout", "-subject"], - capture_output=True, text=True, timeout=5, - ) + r2 = sp.run(["openssl", "x509", "-in", str(path), "-inform", "DER", "-noout", "-subject"], + capture_output=True, text=True, timeout=5) if r2.returncode == 0: info["subject"] = r2.stdout.strip().replace("subject=", "") info["format"] += " (DER)" else: info["subject"] = "UNKNOWN_FORMAT" - info.setdefault("subject", "?") info.setdefault("issuer", "?") info.setdefault("fingerprint", "?") - - # Check if in bundle fp = info.get("fingerprint", "") if fp and fp != "?" and CA_BUNDLE.exists(): with open(CA_BUNDLE) as f: info["in_bundle"] = "yes" if fp in f.read() else "no" else: info["in_bundle"] = "unknown" - return info -def _parse_openssl_output(key: str, rc: int, out: str, err: str) -> None: - """Parse openssl s_client output.""" +def _parse_openssl(key: str, rc: int, out: str, err: str) -> None: for line in out.split("\n"): if "Verify return code" in line: - code = line.strip() - results[key] = code - emoji = "✅" if "0 (ok)" in code else "❌" - print(f" {key}: {emoji} {code}") + results[key] = line.strip() + emoji = "✅" if "0 (ok)" in line else "❌" + print(f" {key}: {emoji} {line.strip()}") return - for line in err.split("\n"): - if line.strip(): - results[key] = f"ERROR: {line.strip()}" - print(f" {key}: ❌ {line.strip()}") - return - results[key] = f"NO_OUTPUT (rc={rc})" - print(f" {key}: ❌ no output") + msg = err.strip() or f"NO_OUTPUT (rc={rc})" + results[key] = f"ERROR: {msg}" + print(f" {key}: ❌ {msg}") + + +def _run(cmd: list[str], timeout: int = 15) -> tuple[int, str, str]: + try: + r = subprocess.run(cmd, 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, "", "timeout" def _make_ctx_cafile(): - """Create SSLContext with cafile (как service.py).""" import ssl return ssl.create_default_context(cafile=str(CA_BUNDLE)) def _make_ctx_cafile_capath(): - """Create SSLContext with cafile + capath.""" import ssl ctx = ssl.create_default_context() if CA_BUNDLE.exists(): @@ -391,112 +137,202 @@ def _make_ctx_cafile_capath(): def _test_httpx(key: str, verify) -> None: - """Test httpx GET with given verify parameter.""" + err = _import("httpx") + if err: + results[key] = err + print(f" {key}: ⚠ {err}") + return import httpx try: r = httpx.get(TARGET_URL, verify=verify, timeout=10) results[key] = f"OK status={r.status_code}" print(f" {key}: ✅ status={r.status_code}") except httpx.ConnectError as e: - s = str(e) - results[key] = f"FAIL ConnectError: {s}" - print(f" {key}: ❌ ConnectError: {s[:200]}") + results[key] = f"FAIL ConnectError" + print(f" {key}: ❌ {e}") except Exception as e: - s = str(e) - results[key] = f"FAIL {type(e).__name__}: {s}" - print(f" {key}: ❌ {type(e).__name__}: {s[:200]}") + results[key] = f"FAIL {type(e).__name__}" + print(f" {key}: ❌ {type(e).__name__}: {e}") def _test_requests(key: str, verify) -> None: - """Test requests GET with given verify parameter.""" + err = _import("requests") + if err: + results[key] = err + print(f" {key}: ⚠ {err}") + return import requests as req try: r = req.get(TARGET_URL, verify=verify, timeout=10) results[key] = f"OK status={r.status_code}" print(f" {key}: ✅ status={r.status_code}") except req.exceptions.SSLError as e: - s = str(e) - results[key] = f"FAIL SSLError: {s}" - print(f" {key}: ❌ SSLError: {s[:200]}") + results[key] = f"FAIL SSLError" + print(f" {key}: ❌ {e}") except Exception as e: - s = str(e) - results[key] = f"FAIL {type(e).__name__}: {s}" - print(f" {key}: ❌ {type(e).__name__}: {s[:200]}") + results[key] = f"FAIL {type(e).__name__}" + print(f" {key}: ❌ {type(e).__name__}: {e}") -def _chain_test() -> None: - """Test with custom chain bundle (all 3 RUSAL certs in one file).""" - # Find the 3 certs - certs = [] +def _section(name: str) -> None: + print(f"\n=== {name} ===") + + +def _diagnose(s: dict) -> str: + issues = [] + for k in ("openssl_default", "openssl_cafile", "openssl_capath"): + v = str(s.get(k, "")) + issues.append(f"{k}={'OK' if '0 (ok)' in v else 'FAIL'}") + for lib in ("httpx", "requests"): + for method in ("true", "cafile", "capath", "false"): + v = str(s.get(f"{lib}_{method}", "")) + issues.append(f"{lib}_{method}={'OK' if v.startswith('OK') else 'FAIL'}") + # Recommendation + if s.get("httpx_cafile_capath", "").startswith("OK") and not s.get("httpx_cafile", "").startswith("OK"): + issues.append("FIX=use_cafile_capath") + elif s.get("requests_capath", "").startswith("OK") and not s.get("requests_cafile", "").startswith("OK"): + issues.append("FIX=use_capath_for_requests") + elif s.get("httpx_false", "").startswith("OK"): + issues.append("FIX=llm_ssl_verify_false_works") + else: + issues.append("FIX=needs_deeper_investigation") + return "; ".join(issues) + + +# ═══════════════════════════════════════════════════════ +# MAIN +# ═══════════════════════════════════════════════════════ + +def main() -> None: + global results + results = {} + host = TARGET_URL.replace("https://", "").split("/")[0] + + _section("1. System Information") + results["target_url"] = TARGET_URL + results["host"] = host + results["llm_ssl_verify"] = LLM_SSL_VERIFY + results["platform"] = platform.platform() + results["python_version"] = sys.version + results["httpx_version"] = _lib_ver("httpx") + results["requests_version"] = _lib_ver("requests") + results["openai_version"] = _lib_ver("openai") + certifi_err = _import("certifi") + if certifi_err: + results["certifi_path"] = certifi_err + results["certifi_certs"] = 0 + else: + import certifi as _certifi + results["certifi_path"] = _certifi.where() + results["certifi_certs"] = _count_certs(_certifi.where()) + print(f" target={TARGET_URL} host={host} platform={platform.platform()}") + + _section("2. System CA Store") + results["ca_bundle"] = str(CA_BUNDLE) + results["ca_bundle_exists"] = CA_BUNDLE.exists() + results["ca_dir_exists"] = CA_DIR.is_dir() + results["ca_bundle_certs"] = _count_certs(str(CA_BUNDLE)) if CA_BUNDLE.exists() else 0 + results["ca_hash_links"] = len(list(CA_DIR.glob("[0-9a-f]*.[0-9]"))) if CA_DIR.is_dir() else 0 + print(f" bundle={CA_BUNDLE} ({results['ca_bundle_certs']} certs)") + print(f" capath={CA_DIR} ({results['ca_hash_links']} hash links)") + print(f" certifi={results.get('certifi_path','?')} ({results.get('certifi_certs','?')} certs)") + print(" custom_certs:") + 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(): + ci = _cert_info(f) + custom_certs.append(ci) + print(f" {ci['file']} subj={ci['subject']} in_bundle={ci['in_bundle']}") + results["custom_certs"] = custom_certs + results["custom_certs_count"] = len(custom_certs) + + _section("3. OpenSSL s_client") + for label, args in [ + ("openssl_default", []), + ("openssl_cafile", ["-CAfile", str(CA_BUNDLE)]), + ("openssl_capath", ["-CApath", str(CA_DIR)]), + ]: + cmd = ["openssl", "s_client", "-connect", f"{host}:443", "-servername", host] + args + rc, out, err = _run(cmd, 15) + _parse_openssl(label, rc, out, err) + + # Chain test: combine all custom certs into one bundle + cert_files = [] for d in CUSTOM_DIRS: if d.is_dir(): for f in d.iterdir(): if f.suffix in (".pem", ".crt"): - certs.append(f) - if len(certs) < 3: - results["openssl_chain"] = f"SKIP (only {len(certs)} certs found)" - print(f" openssl_chain: ⚠ only {len(certs)} certs found, need 3") - return + cert_files.append(f) + if len(cert_files) >= 3: + with tempfile.NamedTemporaryFile(mode="w", suffix=".pem", delete=False) as tf: + chain_path = tf.name + for name in ["RUSAL_ROOT", "Policy_CA", "RGM_Issuing"]: + for f in cert_files: + if name in f.name: + tf.write(f.read_text()) + break + rc, out, err = _run(["openssl", "s_client", "-connect", f"{host}:443", "-CAfile", chain_path, "-servername", host], 15) + os.unlink(chain_path) + _parse_openssl("openssl_chain", rc, out, err) - with tempfile.NamedTemporaryFile(mode="w", suffix=".pem", delete=False) as tf: - chain_path = tf.name - # Составляем в порядке: ROOT → Policy → RGM (сервер → ... → root) - for name in ["RUSAL_ROOT", "Policy_CA", "RGM_Issuing"]: - for f in certs: - if name in f.name: - tf.write(f.read_text()) - break + _section("4. Python httpx (как LLMClient)") + _test_httpx("httpx_true", verify=True) + _test_httpx("httpx_cafile", verify=_make_ctx_cafile()) + _test_httpx("httpx_cafile_capath", verify=_make_ctx_cafile_capath()) + _test_httpx("httpx_false", verify=False) - rc, out, err = run_cmd([ - "openssl", "s_client", "-connect", f"{host}:443", - "-CAfile", chain_path, - "-servername", host, - ], timeout=15) - os.unlink(chain_path) - _parse_openssl_output("openssl_chain", rc, out, err) + _section("5. Python requests (как translate)") + _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) - -def _diagnose(s: dict) -> str: - """Generate AI diagnosis from results.""" - issues = [] - - # openssl - for k in ("openssl_default", "openssl_cafile", "openssl_capath"): - v = s.get(k, "") - if "0 (ok)" in str(v): - issues.append(f"{k}=OK") - else: - issues.append(f"{k}=FAIL") - - # httpx - if s.get("httpx_true", "").startswith("OK"): - issues.append("certifi_works_with_target=yes") + _section("6. NSS Chromium") + rc, out, err = _run(["certutil", "-L", "-d", f"sql:{NSS_DB_DIR}"], 5) + if rc == 0: + results["nss_available"] = True + results["nss_db"] = str(NSS_DB_DIR) + lines = [l.strip() for l in out.split("\n") if l.strip() and "Certificate Nickname" not in l] + results["nss_certs_count"] = len(lines) + print(f" NSS DB: {NSS_DB_DIR} ({len(lines)} certs)") + for l in lines: + if any(x in l.lower() for x in ("rusal", "llm-", "custom-")): + print(f" {l}") else: - issues.append("certifi_works_with_target=no") + results["nss_available"] = False + results["nss_note"] = err.strip() or "certutil or NSS DB unavailable" + print(f" NSS: {results['nss_note']}") - if s.get("httpx_cafile", "").startswith("OK"): - issues.append("system_ca_cafile_works=yes") - else: - issues.append("system_ca_cafile_works=no") + _section("7. LLM_SSL_VERIFY") + raw = LLM_SSL_VERIFY.strip().lower() + disabled = raw in ("false", "0", "no", "off") + results["llm_ssl_verify_disabled"] = disabled + print(f" LLM_SSL_VERIFY={raw} {'DISABLED' if disabled else 'ENABLED'}") + if not disabled: + try: + ctx = _make_ctx_cafile() + results["ssl_context_works"] = True + print(f" SSLContext(cafile) → OK verify_mode={ctx.verify_mode}") + except Exception as e: + results["ssl_context_works"] = False + results["ssl_context_error"] = str(e) + print(f" SSLContext(cafile) → FAIL: {e}") - if s.get("httpx_cafile_capath", "").startswith("OK"): - issues.append("system_ca_cafile_capath_works=yes") - else: - issues.append("system_ca_cafile_capath_works=no") + _section("8. AI Diagnosis Summary") + summary = {k: results[k] for k in [ + "target_url", "host", "llm_ssl_verify", "llm_ssl_verify_disabled", + "ca_bundle_certs", "ca_hash_links", "custom_certs_count", + "openssl_default", "openssl_cafile", "openssl_capath", "openssl_chain", + "httpx_true", "httpx_cafile", "httpx_cafile_capath", "httpx_false", + "requests_true", "requests_cafile", "requests_capath", "requests_false", + "nss_available", "nss_certs_count", "ssl_context_works", + ] if k in results} + summary["ai_diagnosis"] = _diagnose(summary) + print(json.dumps(summary, indent=2, ensure_ascii=False)) - if s.get("httpx_false", "").startswith("OK"): - issues.append("verify_disabled_works=yes") - else: - issues.append("verify_disabled_works=no") - # Which fix to apply - if s.get("httpx_cafile_capath", "").startswith("OK") and not s.get("httpx_cafile", "").startswith("OK"): - issues.append("FIX: use cafile+capath instead of cafile alone") - elif s.get("requests_capath", "").startswith("OK") and not s.get("requests_cafile", "").startswith("OK"): - issues.append("FIX: use capath instead of cafile for requests") - elif s.get("httpx_false", "").startswith("OK"): - issues.append("FIX: LLM_SSL_VERIFY=false works, certs not trusted via system CA") - else: - issues.append("FIX: unknown — need deeper investigation") - - return "; ".join(issues) +if __name__ == "__main__": + main()