busya/ss-tools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: check_llm_certs.py — handle DER files, openssl timeout 20s, binary PEM reads

Browse Source
This commit is contained in:
busya
2026-05-28 23:10:18 +03:00
parent 05f42d5489
commit 8a815ab1b1
1 changed files with 45 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

80
scripts/check_llm_certs.py
View File

@@ -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)