diff --git a/scripts/check_llm_certs.py b/scripts/check_llm_certs.py index 011e28dd..6ff870bd 100755 --- a/scripts/check_llm_certs.py +++ b/scripts/check_llm_certs.py @@ -60,8 +60,8 @@ def _lib_ver(name: str) -> str: def _count_certs(path: str) -> int: try: - with open(path) as f: - return f.read().count("-----BEGIN CERTIFICATE-----") + with open(path, "rb") as f: + return f.read().count(b"-----BEGIN CERTIFICATE-----") except Exception: return 0 @@ -69,31 +69,31 @@ def _count_certs(path: str) -> int: def _cert_info(path: Path) -> dict: 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: + + def _try_read(fmt: str) -> dict | None: + args = ["openssl", "x509", "-in", str(path), "-noout", + "-subject", "-issuer", "-fingerprint", "-sha256"] + if fmt == "DER": + args += ["-inform", "DER"] + r = sp.run(args, capture_output=True, text=True, timeout=5) + if r.returncode != 0: + return None + res = {} 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: - 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", "?") - fp = info.get("fingerprint", "") + if line.startswith("subject="): res["subject"] = line[8:] + elif line.startswith("issuer="): res["issuer"] = line[7:] + elif line.startswith("SHA256 Fingerprint="): res["fingerprint"] = line[19:] + return res + + d = _try_read("PEM") or _try_read("DER") or {} + 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 "" + fp = info["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" + with open(CA_BUNDLE, "rb") as f: + info["in_bundle"] = "yes" if fp.encode() in f.read() else "no" else: info["in_bundle"] = "unknown" return info @@ -118,7 +118,7 @@ def _run(cmd: list[str], timeout: int = 15) -> tuple[int, str, str]: except FileNotFoundError: return -1, "", "command not found" except subprocess.TimeoutExpired: - return -2, "", "timeout" + return -2, "", f"timeout ({timeout}s)" def _make_ctx_cafile(): @@ -184,7 +184,7 @@ def _diagnose(s: dict) -> str: 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"): + for method in ("true", "cafile", "cafile_capath", "capath", "false"): v = str(s.get(f"{lib}_{method}", "")) issues.append(f"{lib}_{method}={'OK' if v.startswith('OK') else 'FAIL'}") # Recommendation @@ -250,33 +250,43 @@ def main() -> None: 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 [ ("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) + cmd = base_cmd + args + rc, out, err = _run(cmd, 20) _parse_openssl(label, rc, out, err) - # Chain test: combine all custom certs into one bundle + # Chain test: combine PEM certs into one bundle (skip DER files) cert_files = [] for d in CUSTOM_DIRS: if d.is_dir(): for f in d.iterdir(): - if f.suffix in (".pem", ".crt"): - cert_files.append(f) + 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) if len(cert_files) >= 3: - with tempfile.NamedTemporaryFile(mode="w", suffix=".pem", delete=False) as tf: + with tempfile.NamedTemporaryFile(mode="wb", 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()) + tf.write(f.read_bytes()) break - rc, out, err = _run(["openssl", "s_client", "-connect", f"{host}:443", "-CAfile", chain_path, "-servername", host], 15) + rc, out, err = _run(["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") _section("4. Python httpx (как LLMClient)") _test_httpx("httpx_true", verify=True)