"""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] }, }