From ed4ac88c3dca558a9a283001da38bb59365bde12 Mon Sep 17 00:00:00 2001 From: busya Date: Thu, 28 May 2026 22:57:08 +0300 Subject: [PATCH] =?UTF-8?q?feat:=20rewrite=20SSL=20diag=20in=20Python=20?= =?UTF-8?q?=E2=80=94=20tests=20all=20real=20libraries?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Tests all 4 SSL methods with actual project libraries: - httpx (verify=True/False/SSLContext cafile/cafile+capath) - requests (verify=True/False/cafile/capath) - openssl s_client (default/CAfile/CApath/chain bundle) - certutil (NSS/Chromium) - certifi vs system CA comparison Outputs JSON summary with AI diagnosis hint. --- scripts/check_llm_certs.py | 502 +++++++++++++++++++++++++++++++++++++ scripts/check_llm_certs.sh | 157 ------------ 2 files changed, 502 insertions(+), 157 deletions(-) create mode 100755 scripts/check_llm_certs.py delete mode 100755 scripts/check_llm_certs.sh diff --git a/scripts/check_llm_certs.py b/scripts/check_llm_certs.py new file mode 100755 index 00000000..74acaafb --- /dev/null +++ b/scripts/check_llm_certs.py @@ -0,0 +1,502 @@ +#!/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] +# @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. +""" + +import json +import os +import platform +import subprocess +import sys +import tempfile +from pathlib import Path + +# ── Configuration ── +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") +CA_DIR = Path("/etc/ssl/certs") +CUSTOM_DIRS = [ + Path("/usr/local/share/ca-certificates/custom"), + Path("/usr/local/share/ca-certificates/llm"), +] +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} ===") + + +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).""" + 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" + + +# ═══════════════════════════════════════════════════════ +# 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 + +# 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-----") + except Exception: + return 0 + + +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"], + 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:] + else: + # Try DER format + 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.""" + 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}") + 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") + + +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(): + ctx.load_verify_locations(cafile=str(CA_BUNDLE)) + if CA_DIR.is_dir(): + ctx.load_verify_locations(capath=str(CA_DIR)) + return ctx + + +def _test_httpx(key: str, verify) -> None: + """Test httpx GET with given verify parameter.""" + 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]}") + except Exception as e: + s = str(e) + results[key] = f"FAIL {type(e).__name__}: {s}" + print(f" {key}: ❌ {type(e).__name__}: {s[:200]}") + + +def _test_requests(key: str, verify) -> None: + """Test requests GET with given verify parameter.""" + 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]}") + except Exception as e: + s = str(e) + results[key] = f"FAIL {type(e).__name__}: {s}" + print(f" {key}: ❌ {type(e).__name__}: {s[:200]}") + + +def _chain_test() -> None: + """Test with custom chain bundle (all 3 RUSAL certs in one file).""" + # Find the 3 certs + certs = [] + 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 + + 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 + + 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) + + +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") + else: + issues.append("certifi_works_with_target=no") + + if s.get("httpx_cafile", "").startswith("OK"): + issues.append("system_ca_cafile_works=yes") + else: + issues.append("system_ca_cafile_works=no") + + 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") + + 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) diff --git a/scripts/check_llm_certs.sh b/scripts/check_llm_certs.sh deleted file mode 100755 index fe4592bf..00000000 --- a/scripts/check_llm_certs.sh +++ /dev/null @@ -1,157 +0,0 @@ -#!/usr/bin/env bash -# #region check_llm_certs [C:4] [TYPE Module] [SEMANTICS ssl, certs, diagnostics, openssl, nss] -# @BRIEF Diagnóstico SSL pro ss-tools. Testuje všechny vrstvy: OpenSSL, -# Python httpx (jako LLMClient), Python requests (jako translate), -# NSS Chromium (jako Playwright). Výstup použije AI k opravě kódu. -# @USAGE ./scripts/check_llm_certs.sh [--target https://...] [-v] -# @LAYER Infrastructure -# @EXAMPLE: -# ./scripts/check_llm_certs.sh --target https://lite.ai.rusal.com -# #endregion check_llm_certs - -set -euo pipefail - -RED='\033[0;31m'; GREEN='\033[0;32m'; YELLOW='\033[1;33m'; CYAN='\033[0;36m'; NC='\033[0m' -PASS="${GREEN}✅${NC}"; FAIL="${RED}❌${NC}"; WARN="${YELLOW}⚠${NC}"; INFO="${CYAN}ℹ${NC}" - -TARGET_URL="${1:-https://lite.ai.rusal.com}" -HOST="$(echo "$TARGET_URL" | sed 's|https://||;s|/.*$||')" -CA_BUNDLE="/etc/ssl/certs/ca-certificates.crt" -CA_DIR="/etc/ssl/certs" -NSS_DB="sql:${HOME:-/root}/.pki/nssdb" - -echo -e "${CYAN}╔════════════════════════════════════════════════════╗${NC}" -echo -e "${CYAN}║ LLM SSL Certificate Diagnostic Report ║${NC}" -echo -e "${CYAN}║ Target: ${TARGET_URL}${NC}" -echo -e "${CYAN}╚════════════════════════════════════════════════════╝${NC}" -echo "" - -# ────────────────────────────────────────────── -# 1. SYSTEM CA STORE -# ────────────────────────────────────────────── -echo -e "${CYAN}── 1. System CA Store ──${NC}" -echo "ca_bundle=${CA_BUNDLE}" -echo "ca_dir=${CA_DIR}" -echo "total_in_bundle=$(grep -c 'BEGIN CERTIFICATE' "$CA_BUNDLE" 2>/dev/null || echo 0)" -echo "total_hash_links=$(find "$CA_DIR" -maxdepth 1 -type l -name '[0-9a-f]*.[0-9]' 2>/dev/null | wc -l)" - -echo "" -echo "custom_certs=" -for dir in /usr/local/share/ca-certificates/custom /usr/local/share/ca-certificates/llm; do - [ -d "$dir" ] || continue - for f in "$dir"/*.pem "$dir"/*.crt; do - [ -f "$f" ] || continue - subj="$(openssl x509 -in "$f" -noout -subject 2>/dev/null | head -1 || echo '?')" - issuer="$(openssl x509 -in "$f" -noout -issuer 2>/dev/null | head -1 || echo '?')" - fp="$(openssl x509 -in "$f" -noout -fingerprint -sha256 2>/dev/null | sed 's/.*=//' || echo '?')" - in_bundle="no" - grep -qF "$fp" "$CA_BUNDLE" 2>/dev/null && in_bundle="yes" - echo " $(basename $f) | subject=${subj} | issuer=${issuer} | in_bundle=${in_bundle}" - done -done - -# ────────────────────────────────────────────── -# 2. OPENSSL s_client - 3 варианта -# ────────────────────────────────────────────── -echo "" -echo -e "${CYAN}── 2. OpenSSL s_client ──${NC}" - -for variant in "CAfile=${CA_BUNDLE}" "CApath=${CA_DIR}" "default"; do - echo "variant=${variant}" - if [ "$variant" = "default" ]; then - out=$(openssl s_client -connect "${HOST}:443" -servername "$HOST" < /dev/null 2>&1) - else - key=$(echo "$variant" | cut -d= -f1) - val=$(echo "$variant" | cut -d= -f2-) - out=$(openssl s_client -connect "${HOST}:443" -"$key" "$val" -servername "$HOST" < /dev/null 2>&1) - fi - code=$(echo "$out" | grep "Verify return code" | head -1) - echo " result=${code}" -done - -# ────────────────────────────────────────────── -# 3. Python - httpx (LLMClient style) -# ────────────────────────────────────────────── -echo "" -echo -e "${CYAN}── 3. Python httpx (как LLMClient) ──${NC}" - -python3 -u -c ' -import os, sys, ssl, traceback - -target = os.environ.get("TARGET_URL", "https://lite.ai.rusal.com") -ca_bundle = "/etc/ssl/certs/ca-certificates.crt" -ca_dir = "/etc/ssl/certs" - -results = {} - -# A) SSLContext s cafile -try: - ctx = ssl.create_default_context(cafile=ca_bundle) - import httpx - r = httpx.get(target, verify=ctx, timeout=10) - results["httpx+cafile"] = f"OK status={r.status_code}" -except Exception as e: - results["httpx+cafile"] = f"FAIL {type(e).__name__}: {e}" - -# B) SSLContext s cafile + capath -try: - ctx2 = ssl.create_default_context() - ctx2.load_verify_locations(cafile=ca_bundle) - if os.path.isdir(ca_dir): - ctx2.load_verify_locations(capath=ca_dir) - import httpx - r = httpx.get(target, verify=ctx2, timeout=10) - results["httpx+cafile+capath"] = f"OK status={r.status_code}" -except Exception as e: - results["httpx+cafile+capath"] = f"FAIL {type(e).__name__}: {e}" - -# C) requests s verify=cafile -try: - import requests - r = requests.get(target, verify=ca_bundle, timeout=10) - results["requests+cafile"] = f"OK status={r.status_code}" -except Exception as e: - results["requests+cafile"] = f"FAIL {type(e).__name__}: {e}" - -# D) requests s verify=capath -try: - import requests - r = requests.get(target, verify=ca_dir, timeout=10) - results["requests+capath"] = f"OK status={r.status_code}" -except Exception as e: - results["requests+capath"] = f"FAIL {type(e).__name__}: {e}" - -for k, v in results.items(): - print(f" {k}={v}") -' 2>&1 - -# ────────────────────────────────────────────── -# 4. NSS / Chromium -# ────────────────────────────────────────────── -echo "" -echo -e "${CYAN}── 4. NSS Chromium ──${NC}" - -if command -v certutil &>/dev/null && certutil -L -d "$NSS_DB" >/dev/null 2>&1; then - echo "nss_db=${NSS_DB}" - echo "nss_certs_count=$(certutil -L -d "$NSS_DB" 2>/dev/null | wc -l)" - echo "nss_custom_certs=" - certutil -L -d "$NSS_DB" 2>/dev/null | grep -iE "rusal|llm-|custom-" | while read -r line; do - echo " $line" - done -else - echo "nss_db=unavailable" -fi - -# ────────────────────────────────────────────── -# 5. LLM_SSL_VERIFY env -# ────────────────────────────────────────────── -echo "" -echo -e "${CYAN}── 5. LLM_SSL_VERIFY ──${NC}" -raw="${LLM_SSL_VERIFY:-not_set}" -echo "LLM_SSL_VERIFY=${raw}" - -# doporuceni -echo "" -echo -e "${CYAN}── 6. AI Recommendation Input ──${NC}" -echo "To fix SSL issues, AI needs to know which methods work." -echo "Copy the FULL output above and paste to the AI assistant."