dennisthiessen
06cf054f60
Extract app/ssl_bootstrap.py (shared with FastAPI main), wire it into research scripts, and teach run_tier1_macbook.sh to locate combined-ca-bundle.pem, certifi, optional USE_CORP_PROXY, plus --ssl-check diagnostics.
137 lines
4.0 KiB
Python
137 lines
4.0 KiB
Python
"""TLS / corporate-proxy bootstrap for CLI scripts and the API.
|
|
|
|
Must run **before** httpx / alpaca / aiohttp open connections.
|
|
|
|
Resolution order for the CA bundle:
|
|
1. ``combined-ca-bundle.pem`` in the repo root (gitignored corporate bundle)
|
|
2. ``$HOME/combined-ca-bundle.pem`` (MacBook path used by existing tooling)
|
|
3. ``SSL_CERT_FILE`` / ``REQUESTS_CA_BUNDLE`` if already set and present
|
|
4. ``certifi.where()`` when the package is installed
|
|
5. System defaults (no patch)
|
|
|
|
Optional corporate proxy (Swisscom-style) when ``USE_CORP_PROXY=1``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import ssl
|
|
from pathlib import Path
|
|
|
|
_BOOTSTRAPPED = False
|
|
|
|
|
|
def _candidate_ca_paths() -> list[Path]:
|
|
root = Path(__file__).resolve().parent.parent
|
|
home = Path.home()
|
|
env_paths = [
|
|
os.environ.get("SSL_CERT_FILE", ""),
|
|
os.environ.get("REQUESTS_CA_BUNDLE", ""),
|
|
os.environ.get("CURL_CA_BUNDLE", ""),
|
|
]
|
|
paths = [
|
|
root / "combined-ca-bundle.pem",
|
|
home / "combined-ca-bundle.pem",
|
|
*[Path(p) for p in env_paths if p],
|
|
]
|
|
try:
|
|
import certifi
|
|
|
|
paths.append(Path(certifi.where()))
|
|
except Exception:
|
|
pass
|
|
return paths
|
|
|
|
|
|
def resolve_ca_bundle() -> str | None:
|
|
for path in _candidate_ca_paths():
|
|
try:
|
|
if path.is_file() and path.stat().st_size > 0:
|
|
return str(path.resolve())
|
|
except OSError:
|
|
continue
|
|
return None
|
|
|
|
|
|
def apply_corp_proxy_if_requested() -> None:
|
|
if os.environ.get("USE_CORP_PROXY", "0") != "1":
|
|
return
|
|
proxy = os.environ.get("CORP_HTTP_PROXY", "http://aproxy.corproot.net:8080")
|
|
no_proxy = os.environ.get(
|
|
"CORP_NO_PROXY",
|
|
"corproot.net,sharedtcs.net,127.0.0.1,localhost,bix.swisscom.com,swisscom.com",
|
|
)
|
|
os.environ.setdefault("HTTP_PROXY", proxy)
|
|
os.environ.setdefault("HTTPS_PROXY", proxy)
|
|
os.environ.setdefault("NO_PROXY", no_proxy)
|
|
os.environ.setdefault("http_proxy", proxy)
|
|
os.environ.setdefault("https_proxy", proxy)
|
|
os.environ.setdefault("no_proxy", no_proxy)
|
|
|
|
|
|
def bootstrap_ssl(*, force: bool = False) -> str | None:
|
|
"""Install CA env vars + patch ``ssl.create_default_context``.
|
|
|
|
Returns the CA path used, or None if nothing was applied.
|
|
Safe to call multiple times.
|
|
"""
|
|
global _BOOTSTRAPPED
|
|
if _BOOTSTRAPPED and not force:
|
|
return os.environ.get("SSL_CERT_FILE") or None
|
|
|
|
apply_corp_proxy_if_requested()
|
|
|
|
cert_path = resolve_ca_bundle()
|
|
if not cert_path:
|
|
_BOOTSTRAPPED = True
|
|
return None
|
|
|
|
os.environ["SSL_CERT_FILE"] = cert_path
|
|
os.environ["REQUESTS_CA_BUNDLE"] = cert_path
|
|
os.environ["CURL_CA_BUNDLE"] = cert_path
|
|
|
|
original = ssl.create_default_context
|
|
|
|
def _patched(
|
|
purpose=ssl.Purpose.SERVER_AUTH, *, cafile=None, capath=None, cadata=None
|
|
):
|
|
ctx = original(purpose, cafile=cafile, capath=capath, cadata=cadata)
|
|
try:
|
|
ctx.load_verify_locations(cafile=cert_path)
|
|
except Exception:
|
|
pass
|
|
return ctx
|
|
|
|
ssl.create_default_context = _patched # type: ignore[assignment]
|
|
|
|
# aiohttp may cache SSL contexts at import time.
|
|
try:
|
|
import aiohttp.connector as aio_conn
|
|
|
|
for attr in ("_SSL_CONTEXT_VERIFIED", "_SSL_CONTEXT_UNVERIFIED"):
|
|
ctx = getattr(aio_conn, attr, None)
|
|
if ctx is not None:
|
|
try:
|
|
ctx.load_verify_locations(cafile=cert_path)
|
|
except Exception:
|
|
pass
|
|
except ImportError:
|
|
pass
|
|
|
|
_BOOTSTRAPPED = True
|
|
return cert_path
|
|
|
|
|
|
def ssl_status() -> dict:
|
|
"""Diagnostic blob for research scripts / MacBook troubleshooting."""
|
|
ca = resolve_ca_bundle()
|
|
return {
|
|
"ca_bundle": ca,
|
|
"ssl_cert_file_env": os.environ.get("SSL_CERT_FILE"),
|
|
"use_corp_proxy": os.environ.get("USE_CORP_PROXY", "0"),
|
|
"http_proxy": os.environ.get("HTTPS_PROXY") or os.environ.get("HTTP_PROXY"),
|
|
"candidates_exist": {
|
|
str(p): p.is_file() for p in _candidate_ca_paths()[:4]
|
|
},
|
|
}
|