Fix SSL cert verification failures behind enterprise inspection proxies

src/ucode/databricks.py

@@ -14,6 +14,7 @@
 import re
 import shlex
 import shutil
+import ssl
 import subprocess
 import time
 from collections.abc import Callable
@@ -229,6 +230,29 @@ def _log_auth_diagnostics() -> None:
         _debug(f"databrickscfg ({cfg_path})", f"read error: {exc}")
 
 
+@functools.cache
+def _make_ssl_context() -> ssl.SSLContext:
+    """Return an SSL context that trusts the system CA bundle plus any custom CA
+    pointed to by REQUESTS_CA_BUNDLE, CURL_CA_BUNDLE, or SSL_CERT_FILE.
+
+    Enterprise environments often inject a self-signed certificate via an SSL
+    inspection proxy. curl picks it up from the system store automatically;
+    Python's default ssl context doesn't on all platforms. Honoring the same
+    env vars that curl and the `requests` library use lets customers point ucode
+    at their enterprise CA bundle without patching the system Python install."""
+    ctx = ssl.create_default_context()
+    for env_var in ("REQUESTS_CA_BUNDLE", "CURL_CA_BUNDLE", "SSL_CERT_FILE"):
+        ca_bundle = os.environ.get(env_var, "").strip()
+        if ca_bundle and Path(ca_bundle).is_file():
+            try:
+                ctx.load_verify_locations(cafile=ca_bundle)
+            except OSError:
+                continue
+            else:
+                break
+    return ctx
+
+
 def _http_get_retry_delay(retry_after: str | None, retry_index: int) -> float:
     if retry_after is not None:
         try:
@@ -272,7 +296,9 @@ def _http_get_json(
     )
     for attempt in range(max_retries + 1):
         try:
-            with urllib_request.urlopen(request, timeout=timeout) as response:
+            with urllib_request.urlopen(
+                request, timeout=timeout, context=_make_ssl_context()
+            ) as response:
                 body = response.read().decode("utf-8")
             _debug(f"GET {url}", f"HTTP 200, {len(body)} bytes")
             if _debug_enabled():
@@ -349,7 +375,9 @@ def _http_send_json(
         headers["Content-Type"] = "application/json"
     request = urllib_request.Request(url, data=body_bytes, method=method, headers=headers)
     try:
-        with urllib_request.urlopen(request, timeout=timeout) as response:
+        with urllib_request.urlopen(
+            request, timeout=timeout, context=_make_ssl_context()
+        ) as response:
             body = response.read().decode("utf-8")
         _debug(f"{method} {url}", f"HTTP {response.status}, {len(body)} bytes")
         if _debug_enabled():
@@ -420,7 +448,9 @@ def _http_get_bytes(url: str, token: str, *, timeout: int = 10) -> tuple[bytes |
     """
     request = urllib_request.Request(url, headers={"Authorization": f"Bearer {token}"})
     try:
-        with urllib_request.urlopen(request, timeout=timeout) as response:
+        with urllib_request.urlopen(
+            request, timeout=timeout, context=_make_ssl_context()
+        ) as response:
             body = response.read()
         _debug(f"GET {url}", f"HTTP 200, {len(body)} bytes")
         return body, None
@@ -3269,7 +3299,7 @@ def discover_sql_warehouses(
     )
 
     try:
-        with urllib_request.urlopen(request, timeout=20) as response:
+        with urllib_request.urlopen(request, timeout=20, context=_make_ssl_context()) as response:
             payload = json.loads(response.read().decode("utf-8"))
     except urllib_error.HTTPError as exc:
         body = exc.read().decode("utf-8", errors="replace") if exc.fp else ""

tests/test_databricks.py

@@ -2208,6 +2208,42 @@ def test_reason_without_body_is_status_only(self):
         assert reason == "HTTP 404 Not Found"
 
 
+class TestMakeSslContext:
+    @pytest.fixture(autouse=True)
+    def clear_cache(self):
+        db_mod._make_ssl_context.cache_clear()
+        yield
+        db_mod._make_ssl_context.cache_clear()
+
+    def test_context_is_cached(self, monkeypatch):
+        from unittest.mock import Mock
+
+        context = Mock()
+        create_default_context = Mock(return_value=context)
+        monkeypatch.setattr(db_mod.ssl, "create_default_context", create_default_context)
+
+        assert db_mod._make_ssl_context() is context
+        assert db_mod._make_ssl_context() is context
+        create_default_context.assert_called_once_with()
+
+    @pytest.mark.parametrize("load_error", [OSError("permission denied"), db_mod.ssl.SSLError()])
+    def test_failed_bundle_falls_back_to_next_one(self, monkeypatch, load_error):
+        from unittest.mock import Mock
+
+        monkeypatch.setenv("REQUESTS_CA_BUNDLE", "/first.pem")
+        monkeypatch.setenv("CURL_CA_BUNDLE", "/second.pem")
+        monkeypatch.setattr(db_mod.Path, "is_file", lambda self: True)
+        context = Mock()
+        context.load_verify_locations.side_effect = [load_error, None]
+        monkeypatch.setattr(db_mod.ssl, "create_default_context", Mock(return_value=context))
+
+        assert db_mod._make_ssl_context() is context
+        assert context.load_verify_locations.call_args_list == [
+            ((), {"cafile": "/first.pem"}),
+            ((), {"cafile": "/second.pem"}),
+        ]
+
+
 class TestHttpGetJsonRetries:
     @staticmethod
     def _http_error(code: int, message: str, headers: dict[str, str] | None = None):
@@ -2225,7 +2261,7 @@ def test_retries_429_after_retry_after_delay(self, monkeypatch):
         calls = []
         sleeps = []
 
-        def fake_urlopen(request, timeout=None):
+        def fake_urlopen(request, timeout=None, **_kwargs):
             calls.append(request)
             outcome = next(outcomes)
             if isinstance(outcome, Exception):
@@ -2249,7 +2285,7 @@ def test_retries_network_error_with_exponential_backoff(self, monkeypatch):
         outcomes = iter([URLError("connection reset"), _FakeResponse({"ok": True})])
         sleeps = []
 
-        def fake_urlopen(request, timeout=None):
+        def fake_urlopen(request, timeout=None, **_kwargs):
             outcome = next(outcomes)
             if isinstance(outcome, Exception):
                 raise outcome
@@ -2269,7 +2305,7 @@ def test_stops_after_configured_retries(self, monkeypatch):
         calls = []
         sleeps = []
 
-        def fake_urlopen(request, timeout=None):
+        def fake_urlopen(request, timeout=None, **_kwargs):
             calls.append(request)
             raise self._http_error(429, "Too Many Requests")
 
@@ -2287,7 +2323,7 @@ def fake_urlopen(request, timeout=None):
     def test_does_not_retry_other_http_error(self, monkeypatch):
         calls = []
 
-        def fake_urlopen(request, timeout=None):
+        def fake_urlopen(request, timeout=None, **_kwargs):
             calls.append(request)
             raise self._http_error(503, "Service Unavailable")
 
@@ -2537,7 +2573,7 @@ class TestHttpGetJsonTimeout:
     escapes the best-effort MCP discovery flow and crashes the command."""
 
     def test_read_timeout_returns_reason_instead_of_raising(self, monkeypatch):
-        def raise_timeout(request, timeout=None):
+        def raise_timeout(request, timeout=None, **_kwargs):
             raise TimeoutError("The read operation timed out")
 
         monkeypatch.setattr(db_mod.urllib_request, "urlopen", raise_timeout)
@@ -2549,7 +2585,7 @@ def raise_timeout(request, timeout=None):
         assert "timed out" in reason
 
     def test_post_read_timeout_returns_reason_instead_of_raising(self, monkeypatch):
-        def raise_timeout(request, timeout=None):
+        def raise_timeout(request, timeout=None, **_kwargs):
             raise TimeoutError("The read operation timed out")
 
         monkeypatch.setattr(db_mod.urllib_request, "urlopen", raise_timeout)
@@ -2879,7 +2915,7 @@ def _empty_response(body: str = ""):
     def test_empty_body_is_success_not_a_decode_error(self, monkeypatch):
         # Without `allow_empty_body` this would fail with "response was not valid JSON".
         monkeypatch.setattr(
-            db_mod.urllib_request, "urlopen", lambda request, timeout=None: self._empty_response()
+            db_mod.urllib_request, "urlopen", lambda request, timeout=None, **_k: self._empty_response()
         )
         payload, reason = db_mod._http_delete(f"{WS}/api/anything", "tok")
         assert reason is None
@@ -2889,7 +2925,7 @@ def test_empty_json_object_is_also_success(self, monkeypatch):
         monkeypatch.setattr(
             db_mod.urllib_request,
             "urlopen",
-            lambda request, timeout=None: self._empty_response("{}"),
+            lambda request, timeout=None, **_k: self._empty_response("{}"),
         )
         payload, reason = db_mod._http_delete(f"{WS}/api/anything", "tok")
         assert reason is None
@@ -2898,7 +2934,7 @@ def test_empty_json_object_is_also_success(self, monkeypatch):
     def test_uses_the_delete_verb_and_sends_no_body(self, monkeypatch):
         seen = {}
 
-        def capture(request, timeout=None):
+        def capture(request, timeout=None, **_kwargs):
             seen["method"] = request.get_method()
             seen["data"] = request.data
             return self._empty_response()
@@ -2915,7 +2951,7 @@ def test_http_error_surfaces_the_body(self, monkeypatch):
 
         body = '{"error_code":"PERMISSION_DENIED","message":"admin required"}'
 
-        def raise_http_error(request, timeout=None):
+        def raise_http_error(request, timeout=None, **_kwargs):
             raise HTTPError(
                 url="", code=403, msg="Forbidden", hdrs=MagicMock(), fp=io.BytesIO(body.encode())
             )
@@ -2933,7 +2969,7 @@ def test_uses_the_patch_verb_and_sends_the_body(self, monkeypatch):
 
         seen = {}
 
-        def capture(request, timeout=None):
+        def capture(request, timeout=None, **_kwargs):
             seen["method"] = request.get_method()
             seen["data"] = request.data
             seen["content_type"] = request.get_header("Content-type")