mirror of
https://github.com/pewdiepie-archdaemon/odysseus.git
synced 2026-10-22 14:52:24 +02:00
181 lines
8.8 KiB
Python
181 lines
8.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Reject the eMudhra incident's known CA certificates in PEM trust stores.
|
|
|
|
This checks certificate contents, including OpenSSL's lazily loaded CA directory.
|
|
It does not model native OS certificate revocation or arbitrary custom clients.
|
|
Incident fingerprints: https://emudhra.we-have-a-problem.com/roots.json
|
|
and https://emudhra.we-have-a-problem.com/impacted-intermediates.json
|
|
Cross-signed issuer: https://emudhra.we-have-a-problem.com/chains.zip
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import importlib
|
|
import importlib.util
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import ssl
|
|
import sys
|
|
|
|
|
|
BLOCKED_CERTIFICATES = {
|
|
"86a1ecba089c4a8d3bbe2734c612ba341d813e043cf9e8a862cd5c57a36bbe6b": "emSign ECC Root CA - G3",
|
|
"d7a7a0fb5d7e2731d771e9484ebcdef71d5f0c3e0a2948782bc83ee0ea699ef4": "AAA Certificate Services",
|
|
"7dd78d5f4f13459a83dff9abbba62edbaf6f2d102bf257fd712f4d9f2746ed8d": "emSign Root TLS CA - G3",
|
|
"f2edaaff3d9b701fb4d14c9f285f8c0bad90fc3bcc1fa07d47f4718edaba34a5": "emSign Root CS CA - G3",
|
|
"3094025792039a68af6e553686410bdab7c0d4a36b9084cba64202e35d6ebbdd": "emSign Root TLS CA - G3 (cross-signed)",
|
|
# Intermediates can themselves be operator-supplied trust anchors. Include
|
|
# both published signatures for each key, without parsing SPKI in stdlib.
|
|
"42c4accd12c4f59597925e1a08a7a90102608543809308d50ec427e118ba1493": "EM DV TLS CA - G3B (eMudhra)",
|
|
"45a883f349aedfd5f3a649b5498ffb121712ba8b448f3f2a91dbc5a766fcddbe": "EM DV TLS CA - G3B (Sectigo)",
|
|
"98b896625c0c230adeec3eb2cf6aad5001ecf9156a68644e82374ce5833bc25b": "EM OV TLS CA - G3B (eMudhra)",
|
|
"5f1ded4dd590af74631290cf073b122a2aac6a1fa479778a0bd43579bd832699": "EM OV TLS CA - G3B (Sectigo)",
|
|
"e7568fe6e12331f4fd7de61da4c7559d0f8eb192a218d4741a3ba75184333838": "EM EV TLS CA - G3B (eMudhra)",
|
|
"579480aea166290bd3c643589e795fc0d0e54e86505d180f1dff99efd3199512": "EM EV TLS CA - G3B (Sectigo)",
|
|
"0164d7ceae5bb909c272af151ff44b1177ae9b47d9424b94f1d9558c30934d19": "EM DV TLS CA - G3C (eMudhra)",
|
|
"62b7f9868263f72f6911fc78ceaf72c1979158b7368bf648683bb2aa39b02e72": "EM DV TLS CA - G3C (Sectigo)",
|
|
"cb6e15165ac802e9b3f993f16c25d41f4e81c21ebde71ae1b012ac00bf9c7c3f": "EM OV TLS CA - G3C (eMudhra)",
|
|
"7485b51c0ed192878fa1d2f948fb529ae1842cc5086c4b67f7be5b2039de9489": "EM OV TLS CA - G3C (Sectigo)",
|
|
"432ec25f0d28de9006d97bf752f17bbb2cd41f669933fcb072d9a1148d0261f3": "EM EV TLS CA - G3C (eMudhra)",
|
|
"fd9d5c2dbbd4995ab69b8d50c439d9527b00529199c2fd01b2174eb58f07021f": "EM EV TLS CA - G3C (Sectigo)",
|
|
"9e39de2881273acc225a149c72ba3e0f51424963669ab633ee1dc7df5db067b4": "EM OV CS CA - G3B (eMudhra)",
|
|
"a496e23ac0cab8155fc9b117b97eae368efcbb9041aec8804d5a308d2d01c2d8": "EM OV CS CA - G3B (Sectigo)",
|
|
"4b2460a466f95fd40113bcdcca2ceb29b3b093921de45474da7df48874c630b1": "EM OV CS CA - G3C (eMudhra)",
|
|
"84112972b8e59d14560ae7e7119654474aad119990cf3722bfc1e3b91615e1fd": "EM OV CS CA - G3C (Sectigo)",
|
|
}
|
|
OVERRIDE_ENV_VARS = (
|
|
"LLM_CA_BUNDLE", "SSL_CERT_FILE", "REQUESTS_CA_BUNDLE", "CURL_CA_BUNDLE",
|
|
"PIP_CERT", "NODE_EXTRA_CA_CERTS", "GRPC_DEFAULT_SSL_ROOTS_FILE_PATH",
|
|
)
|
|
|
|
|
|
def check_bundle(path: Path) -> None:
|
|
"""Raise on an unreadable, empty, malformed, or affected CA bundle."""
|
|
context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
|
|
context.load_verify_locations(cafile=str(path))
|
|
# A hashed OpenSSL directory can include valid non-CA certificates (for
|
|
# example Ubuntu's snakeoil server certificate). get_ca_certs() omits them;
|
|
# use the total X.509 count to distinguish these from an empty store.
|
|
if not context.cert_store_stats()["x509"]:
|
|
raise ValueError("no certificates")
|
|
certificates = context.get_ca_certs(binary_form=True)
|
|
blocked = {
|
|
BLOCKED_CERTIFICATES[fingerprint]
|
|
for certificate in certificates
|
|
if (fingerprint := hashlib.sha256(certificate).hexdigest()) in BLOCKED_CERTIFICATES
|
|
}
|
|
if blocked:
|
|
raise ValueError("affected CA certificates: " + ", ".join(sorted(blocked)))
|
|
|
|
|
|
def ca_directory_paths(directory: Path) -> list[Path]:
|
|
"""Return the hashed certificate files consumed by OpenSSL's CApath."""
|
|
return [
|
|
path for path in directory.iterdir()
|
|
if re.fullmatch(r"[0-9a-fA-F]{8}\.[0-9]+", path.name) and path.is_file()
|
|
]
|
|
|
|
|
|
def system_paths() -> list[Path]:
|
|
"""Inspect both the effective OpenSSL CA file and hashed directory entries."""
|
|
defaults = ssl.get_default_verify_paths()
|
|
paths = [Path(defaults.cafile)] if defaults.cafile else []
|
|
# get_default_verify_paths() can report None for a colon-separated
|
|
# SSL_CERT_DIR even though OpenSSL consumes each directory in that list.
|
|
capath = os.environ.get(defaults.openssl_capath_env, defaults.capath)
|
|
if capath:
|
|
for directory in filter(None, capath.split(os.pathsep)):
|
|
paths.extend(ca_directory_paths(Path(directory)))
|
|
if not paths:
|
|
raise ValueError("no effective OpenSSL PEM CA store found")
|
|
return paths
|
|
|
|
|
|
def override_paths() -> list[Path]:
|
|
"""Inspect every configured override, including client fallback settings."""
|
|
paths = []
|
|
for name in OVERRIDE_ENV_VARS:
|
|
value = os.environ.get(name, "")
|
|
if name == "LLM_CA_BUNDLE":
|
|
value = value.strip() # Match src/tls_overrides.py.
|
|
if not value:
|
|
continue
|
|
path = Path(value)
|
|
# Requests (also used by pip) accepts either a file or a hashed directory.
|
|
if name in ("REQUESTS_CA_BUNDLE", "CURL_CA_BUNDLE", "PIP_CERT") and path.is_dir():
|
|
entries = ca_directory_paths(path)
|
|
if not entries:
|
|
raise ValueError(f"{name}: no hashed certificates in {path}")
|
|
paths.extend(entries)
|
|
else:
|
|
paths.append(path)
|
|
return paths
|
|
|
|
|
|
def grpc_roots_path() -> Path:
|
|
"""Select gRPC's override or its independent Python-wheel default bundle."""
|
|
override = os.environ.get("GRPC_DEFAULT_SSL_ROOTS_FILE_PATH")
|
|
if override:
|
|
# gRPC can fall back to its bundled roots when this file is unreadable.
|
|
# Check the configured file strictly instead of blessing that fallback.
|
|
return Path(override)
|
|
spec = importlib.util.find_spec("grpc")
|
|
if spec is None or spec.origin is None:
|
|
raise ImportError("grpcio is not installed; cannot check its default roots")
|
|
return Path(spec.origin).parent / "_cython" / "_credentials" / "roots.pem"
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("bundles", nargs="*", type=Path)
|
|
parser.add_argument("--system", action="store_true", help="check effective OpenSSL PEM stores")
|
|
parser.add_argument("--python", action="store_true", help="check certifi and pip's separate vendored bundle")
|
|
parser.add_argument("--grpc", action="store_true", help="check gRPC's selected default CA file")
|
|
parser.add_argument("--overrides", action="store_true", help="check configured LLM/HTTPX/Requests/pip/Node/gRPC CA overrides")
|
|
parser.add_argument("--verbose", action="store_true", help="show the runtime identity, Python executable and selected CA paths")
|
|
args = parser.parse_args(argv)
|
|
if not (args.bundles or args.system or args.python or args.grpc or args.overrides):
|
|
parser.error("select --system, --python, --grpc, --overrides, or explicit PEM bundle paths")
|
|
if args.verbose:
|
|
print(f"Python executable: {sys.executable}")
|
|
print(f"Python prefix: {sys.prefix}")
|
|
print(f"User home: {Path.home()}")
|
|
if hasattr(os, "geteuid"):
|
|
print(f"Effective user: uid={os.geteuid()} gid={os.getegid()}")
|
|
try:
|
|
paths = list(args.bundles)
|
|
if args.system:
|
|
paths.extend(system_paths())
|
|
if args.python:
|
|
for name in ("certifi", "pip._vendor.certifi"):
|
|
paths.append(Path(importlib.import_module(name).where()))
|
|
if args.grpc:
|
|
paths.append(grpc_roots_path())
|
|
if args.overrides:
|
|
paths.extend(override_paths())
|
|
paths = list(dict.fromkeys(path.resolve() for path in paths))
|
|
if not paths:
|
|
raise ValueError("no CA bundle files selected; also select --system, --python, or an explicit path")
|
|
errors = []
|
|
for path in paths:
|
|
if args.verbose:
|
|
print(f"CA bundle: {path}")
|
|
try:
|
|
check_bundle(path)
|
|
except (OSError, ValueError, ssl.SSLError) as exc:
|
|
errors.append(f"{path}: {exc}")
|
|
if errors:
|
|
raise ValueError("\n".join(errors))
|
|
except (ImportError, OSError, ValueError, ssl.SSLError) as exc:
|
|
print(str(exc), file=sys.stderr)
|
|
return 1
|
|
print(f"Validated {len(paths)} CA bundle file(s): no known eMudhra incident CA certificates")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|