From 802727b58ae527f93a1bcfa5b0d4a1a60bf264fa Mon Sep 17 00:00:00 2001 From: busya Date: Thu, 28 May 2026 23:22:27 +0300 Subject: [PATCH] =?UTF-8?q?fix:=20check=5Fllm=5Fcerts.py=20=E2=80=94=20fix?= =?UTF-8?q?es=20from=20user=20review?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - openssl s_client: stdin=EOF to prevent hang (input='') - argparse for --target CLI argument - in_bundle check: awk+openssl per-cert fingerprint comparison - NamedTemporaryFile: mode='wb' for binary writes - Added httpx_capath and requests_cafile_capath tests - Graceful handling of empty ca-certificates.crt --- scripts/check_llm_certs.py | 188 ++++++++++++++++++++++++------------- 1 file changed, 121 insertions(+), 67 deletions(-) diff --git a/scripts/check_llm_certs.py b/scripts/check_llm_certs.py index 6ff870bd..b2ece817 100755 --- a/scripts/check_llm_certs.py +++ b/scripts/check_llm_certs.py @@ -1,14 +1,15 @@ #!/usr/bin/env python3 # #region check_llm_certs [C:4] [TYPE Module] [SEMANTICS ssl, certs, diagnostics, python] -# @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. +# @BRIEF SSL diagnostics in Python — tests all real libraries used by ss-tools. +# Reads TARGET_URL from env or CLI --target. +# Uses capath by default (cafile is unreliable — see ADR-0009). # @USAGE ./scripts/check_llm_certs.py [--target https://...] # @LAYER Infrastructure # #endregion check_llm_certs from __future__ import annotations +import argparse import json import os import platform @@ -18,9 +19,7 @@ import tempfile import traceback from pathlib import Path -# ── 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") +# ── Configuration defaults ── CA_BUNDLE = Path("/etc/ssl/certs/ca-certificates.crt") CA_DIR = Path("/etc/ssl/certs") CUSTOM_DIRS = [ @@ -35,9 +34,7 @@ results: dict = {} # ── Lazy library loader ── _LIBS: dict = {} - 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: @@ -45,10 +42,9 @@ def _import(name: str) -> str | None: _LIBS[name] = True return None except ImportError: - _LIBS[name] = f"NOT_INSTALLED" + _LIBS[name] = "NOT_INSTALLED" return _LIBS[name] - def _lib_ver(name: str) -> str: err = _import(name) if err: @@ -57,7 +53,6 @@ def _lib_ver(name: str) -> str: v = getattr(mod, "__version__", None) return str(v) if v else "installed" - def _count_certs(path: str) -> int: try: with open(path, "rb") as f: @@ -66,8 +61,10 @@ def _count_certs(path: str) -> int: return 0 +# ── Certificate info with proper in_bundle check ── + def _cert_info(path: Path) -> dict: - import subprocess as sp + """Extract cert info. Tries PEM first, then DER.""" info = {"file": str(path), "format": path.suffix} def _try_read(fmt: str) -> dict | None: @@ -75,7 +72,7 @@ def _cert_info(path: Path) -> dict: "-subject", "-issuer", "-fingerprint", "-sha256"] if fmt == "DER": args += ["-inform", "DER"] - r = sp.run(args, capture_output=True, text=True, timeout=5) + r = subprocess.run(args, capture_output=True, text=True, timeout=5) if r.returncode != 0: return None res = {} @@ -89,29 +86,43 @@ def _cert_info(path: Path) -> dict: info["subject"] = d.get("subject", "UNKNOWN_FORMAT") info["issuer"] = d.get("issuer", "?") info["fingerprint"] = d.get("fingerprint", "?") - info["format"] += " (DER)" if d is not None and _try_read("PEM") is None else "" + if _try_read("PEM") is None and d: + info["format"] += " (DER)" + + # Proper in_bundle check: compare fingerprint against EACH cert in bundle fp = info["fingerprint"] + info["in_bundle"] = "unknown" if fp and fp != "?" and CA_BUNDLE.exists(): - with open(CA_BUNDLE, "rb") as f: - info["in_bundle"] = "yes" if fp.encode() in f.read() else "no" - else: - info["in_bundle"] = "unknown" + rc, out, _ = _run( + ["awk", '-vcmd=openssl x509 -noout -fingerprint -sha256', + '/BEGIN/{close(cmd)};{print | cmd}', str(CA_BUNDLE)], 10 + ) + if rc == 0: + info["in_bundle"] = "yes" if fp in out else "no" + else: + # Fallback: simple file search + with open(CA_BUNDLE, "rb") as f: + info["in_bundle"] = "yes" if fp.encode() in f.read() else "no" return info -def _parse_openssl(key: str, rc: int, out: str, err: str) -> None: - for line in out.split("\n"): - if "Verify return code" in line: - results[key] = line.strip() - emoji = "✅" if "0 (ok)" in line else "❌" - print(f" {key}: {emoji} {line.strip()}") - return - msg = err.strip() or f"NO_OUTPUT (rc={rc})" - results[key] = f"ERROR: {msg}" - print(f" {key}: ❌ {msg}") +# ── OpenSSL s_client runner (with stdin EOF to prevent hang) ── + +def _run_openssl(cmd: list[str], timeout: int = 20) -> tuple[int, str, str]: + """Run openssl s_client with stdin=EOF to prevent hang after handshake.""" + full_cmd = cmd + ["-verify_return_error"] + try: + r = subprocess.run(full_cmd, input="", 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 _run(cmd: list[str], timeout: int = 15) -> tuple[int, str, str]: + """Generic subprocess runner.""" try: r = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout) return r.returncode, r.stdout, r.stderr @@ -121,21 +132,47 @@ def _run(cmd: list[str], timeout: int = 15) -> tuple[int, str, str]: return -2, "", f"timeout ({timeout}s)" -def _make_ctx_cafile(): +# ── OpenAI-compatible s_client parsing ── + +def _parse_openssl(key: str, rc: int, out: str, err: str) -> None: + for line in out.split("\n"): + if "Verify return code" in line: + results[key] = line.strip() + emoji = "✅" if "0 (ok)" in line else "❌" + print(f" {key}: {emoji} {line.strip()}") + return + # If timed out, say so clearly + if "timeout" in err: + results[key] = f"TIMEOUT (proxy? check HTTP_PROXY env)" + print(f" {key}: ❌ {results[key]}") + return + msg = err.strip() or f"NO_OUTPUT (rc={rc})" + results[key] = f"ERROR: {msg}" + print(f" {key}: ❌ {msg}") + + +# ── SSL context builders ── + +def _ctx_cafile(): import ssl return ssl.create_default_context(cafile=str(CA_BUNDLE)) +def _ctx_capath(): + import ssl + return ssl.create_default_context(capath=str(CA_DIR)) -def _make_ctx_cafile_capath(): +def _ctx_cafile_capath(): import ssl ctx = ssl.create_default_context() - if CA_BUNDLE.exists(): + if CA_BUNDLE.exists() and _count_certs(str(CA_BUNDLE)) > 0: ctx.load_verify_locations(cafile=str(CA_BUNDLE)) if CA_DIR.is_dir(): ctx.load_verify_locations(capath=str(CA_DIR)) return ctx +# ── HTTP testers ── + def _test_httpx(key: str, verify) -> None: err = _import("httpx") if err: @@ -154,7 +191,6 @@ def _test_httpx(key: str, verify) -> None: results[key] = f"FAIL {type(e).__name__}" print(f" {key}: ❌ {type(e).__name__}: {e}") - def _test_requests(key: str, verify) -> None: err = _import("requests") if err: @@ -178,18 +214,25 @@ def _section(name: str) -> None: print(f"\n=== {name} ===") +# ═══════════════════════════════════════════════════════ +# DIAGNOSIS +# ═══════════════════════════════════════════════════════ + 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", "cafile_capath", "capath", "false"): + for lib, methods in [("httpx", ("true", "cafile", "capath", "cafile_capath", "false")), + ("requests", ("true", "cafile", "capath", "false"))]: + for method in methods: 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") + issues.append("FIX=use_cafile_capath_for_httpx") + elif s.get("httpx_capath", "").startswith("OK") and not s.get("httpx_cafile", "").startswith("OK"): + issues.append("FIX=use_capath_for_httpx") 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"): @@ -204,9 +247,16 @@ def _diagnose(s: dict) -> str: # ═══════════════════════════════════════════════════════ def main() -> None: - global results + global results, TARGET_URL results = {} + + parser = argparse.ArgumentParser(description="SSL Certificate Diagnostic Tool") + parser.add_argument("--target", default=os.environ.get("TARGET_URL", "https://lite.ai.rusal.com"), + help="Target URL to test (default: https://lite.ai.rusal.com)") + args = parser.parse_args() + TARGET_URL = args.target host = TARGET_URL.replace("https://", "").split("/")[0] + LLM_SSL_VERIFY = os.environ.get("LLM_SSL_VERIFY", "true") _section("1. System Information") results["target_url"] = TARGET_URL @@ -223,19 +273,20 @@ def main() -> None: 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()}") + certifi_path = _certifi.where() + results["certifi_path"] = certifi_path + results["certifi_certs"] = _count_certs(certifi_path) + print(f" target={TARGET_URL} host={host}") + print(f" httpx={results['httpx_version']} requests={results['requests_version']} openai={results['openai_version']}") + print(f" certifi={results.get('certifi_path','?')} ({results.get('certifi_certs','?')} certs)") _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: @@ -245,31 +296,28 @@ def main() -> None: 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']}") + print(f" {ci['file']}") + print(f" subj={ci['subject']} fmt={ci['format']} in_bundle={ci['in_bundle']}") results["custom_certs"] = custom_certs results["custom_certs_count"] = len(custom_certs) _section("3. OpenSSL s_client") - # Note: if behind a corporate proxy, openssl s_client may hang. - # Adding -verify_return_error and -msg for better diagnostics. - base_cmd = ["openssl", "s_client", "-connect", f"{host}:443", - "-servername", host, "-verify_return_error"] - for label, args in [ + for label, extra_args in [ ("openssl_default", []), ("openssl_cafile", ["-CAfile", str(CA_BUNDLE)]), ("openssl_capath", ["-CApath", str(CA_DIR)]), ]: - cmd = base_cmd + args - rc, out, err = _run(cmd, 20) + cmd = ["openssl", "s_client", "-connect", f"{host}:443", + "-servername", host] + extra_args + rc, out, err = _run_openssl(cmd, 20) _parse_openssl(label, rc, out, err) - # Chain test: combine PEM certs into one bundle (skip DER files) + # Chain test: combine PEM certs cert_files = [] for d in CUSTOM_DIRS: if d.is_dir(): for f in d.iterdir(): if f.suffix in (".pem", ".crt") and f.is_file(): - # Check if PEM rc, _, _ = _run(["openssl", "x509", "-in", str(f), "-noout"], 3) if rc == 0: cert_files.append(f) @@ -281,17 +329,19 @@ def main() -> None: if name in f.name: tf.write(f.read_bytes()) break - rc, out, err = _run(["openssl", "s_client", "-connect", f"{host}:443", - "-CAfile", chain_path, "-servername", host], 20) + rc, out, err = _run_openssl( + ["openssl", "s_client", "-connect", f"{host}:443", + "-CAfile", chain_path, "-servername", host], 20) os.unlink(chain_path) _parse_openssl("openssl_chain", rc, out, err) else: - print(" openssl_chain: ⚠ < 3 PEM certs found, skipping") + print(" openssl_chain: ⚠ < 3 PEM certs, skipping") _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_cafile", verify=_ctx_cafile()) + _test_httpx("httpx_capath", verify=_ctx_capath()) + _test_httpx("httpx_cafile_capath", verify=_ctx_cafile_capath()) _test_httpx("httpx_false", verify=False) _section("5. Python requests (как translate)") @@ -305,9 +355,10 @@ def main() -> None: 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] + 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)") + print(f" DB={NSS_DB_DIR} certs={len(lines)}") for l in lines: if any(x in l.lower() for x in ("rusal", "llm-", "custom-")): print(f" {l}") @@ -316,30 +367,33 @@ def main() -> None: results["nss_note"] = err.strip() or "certutil or NSS DB unavailable" print(f" NSS: {results['nss_note']}") - _section("7. LLM_SSL_VERIFY") + _section("7. LLM_SSL_VERIFY env") 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'}") + print(f" LLM_SSL_VERIFY={raw} {'DISABLED' if disabled else 'ENABLED'}") if not disabled: try: - ctx = _make_ctx_cafile() + ctx = _ctx_cafile() results["ssl_context_works"] = True - print(f" SSLContext(cafile) → OK verify_mode={ctx.verify_mode}") + print(f" SSLContext(cafile) works, 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}") + print(f" SSLContext(cafile) FAIL: {e}") _section("8. AI Diagnosis Summary") - summary = {k: results[k] for k in [ + summary = {} + 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", + "httpx_true", "httpx_cafile", "httpx_capath", "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} + ]: + if k in results: + summary[k] = results[k] summary["ai_diagnosis"] = _diagnose(summary) print(json.dumps(summary, indent=2, ensure_ascii=False))