Allow setting REQUESTS_CA_BUNDLE to use custom certs for OpenAI embedders (#561)

If your embedding models are hosted within an enterprise/private PKI
environment, you many need to trust a custom certificate bundle when
interacting with them. This change allows setting REQUESTS_CA_BUNDLE to
a file which will be used by the HTTP client in OpenAI and Azure OpenAI
embedding interactions.

---------

Co-authored-by: ryannikolaidis <[email protected]>

Author: ctrahey

CHANGELOG.md

@@ -1,3 +1,7 @@
+## 1.1.0
+
+- **Feature**: Embedding with OpenAI (or Azure OpenAI) can trust custom certificate authority by specifying environment variable REQUESTS_CA_BUNDLE.
+
 ## 1.0.59
 
 * **o11y: Downgrade OTEL logs to `DEBUG` by default, make it configurable**

requirements/base.txt

@@ -4,3 +4,4 @@ pydantic>=2.7
 tqdm
 click
 opentelemetry-sdk
+certifi>=2025.7.14

test/integration/embedders/conftest.py

@@ -1,6 +1,19 @@
+import http.server
+import json
+import os
+import ssl
+import tempfile
+import threading
+from collections import Counter
+from datetime import datetime, timedelta, timezone
 from pathlib import Path
+from typing import Generator
 
 import pytest
+from cryptography import x509
+from cryptography.hazmat.primitives import hashes, serialization
+from cryptography.hazmat.primitives.asymmetric import rsa
+from cryptography.x509.oid import NameOID
 
 
 @pytest.fixture
@@ -11,3 +24,76 @@ def embedder_file() -> Path:
     assert embedder_file.exists()
     assert embedder_file.is_file()
     return embedder_file
+
+
+_EMBEDDINGS_CALLS = Counter()
+
+
+class MockOpenAIEmbeddingsHandler(http.server.SimpleHTTPRequestHandler):
+    """
+    Minimal OpenAPI Completions mock server
+    """
+
+    def do_POST(self):
+        global _EMBEDDINGS_CALLS
+        _EMBEDDINGS_CALLS["POST"] += 1
+        self.send_response(200)
+        self.send_header("Content-type", "application/json")
+        self.end_headers()
+        body = {
+            "data": [{"object": "embedding", "embedding": [], "index": 0}],
+            "object": "list",
+            "model": "text-embedding-ada-002",
+            "usage": {"prompt_tokens": 1, "total_tokens": 2},
+        }
+        self.wfile.write(json.dumps(body).encode("utf-8"))
+
+
+@pytest.fixture(scope="module")
+def mock_embeddings_server() -> Generator[tuple[int, str, Counter], None, None]:
+    """
+    Runs a dead-simple HTTPS server on a random port, in a thread, with a custom TLS certificate.
+    """
+
+    private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+    subject = issuer = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "localhost")])
+    cert = (
+        x509.CertificateBuilder()
+        .subject_name(subject)
+        .issuer_name(issuer)
+        .public_key(private_key.public_key())
+        .serial_number(x509.random_serial_number())
+        .not_valid_before(datetime.now(timezone.utc))
+        .not_valid_after(datetime.now(timezone.utc) + timedelta(days=1))
+        .sign(private_key, hashes.SHA256())
+    )
+
+    with (
+        tempfile.NamedTemporaryFile(delete=False, suffix=".pem") as cert_file,
+        tempfile.NamedTemporaryFile(delete=False, suffix=".pem") as key_file,
+    ):
+        cert_file.write(cert.public_bytes(serialization.Encoding.PEM))
+        key_file.write(
+            private_key.private_bytes(
+                encoding=serialization.Encoding.PEM,
+                format=serialization.PrivateFormat.PKCS8,
+                encryption_algorithm=serialization.NoEncryption(),
+            )
+        )
+        cert_fpath = cert_file.name
+        privkey_fpath = key_file.name
+
+    context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+    context.load_cert_chain(certfile=cert_fpath, keyfile=privkey_fpath, password="")
+    server_address = "127.0.0.1", 0
+
+    httpd = http.server.HTTPServer(server_address, MockOpenAIEmbeddingsHandler)
+    httpd.socket = context.wrap_socket(httpd.socket, server_side=True)
+    thread = threading.Thread(target=httpd.serve_forever)
+    thread.daemon = True
+    thread.start()
+    yield httpd.server_port, cert_fpath, _EMBEDDINGS_CALLS
+    httpd.shutdown()
+    thread.join()
+    os.unlink(cert_fpath)
+    os.unlink(privkey_fpath)

test/integration/embedders/test_azure_openai.py

@@ -3,6 +3,9 @@
 from dataclasses import dataclass
 from pathlib import Path
 
+import pydantic
+import pytest
+
 from test.integration.embedders.utils import validate_embedding_output, validate_raw_embedder
 from test.integration.utils import requires_env
 from unstructured_ingest.embed.azure_openai import (
@@ -57,3 +60,43 @@ def test_raw_azure_openai_embedder(embedder_file: Path):
     )
     embedder.precheck()
     validate_raw_embedder(embedder=embedder, embedder_file=embedder_file, expected_dimension=1536)
+
+
+def test_openai_custom_tls_no_override_should_fail(mock_embeddings_server, embedder_file: Path):
+    from openai import APIConnectionError
+
+    port, certificate_path, counter = mock_embeddings_server
+    calls_before = counter["POST"]
+    with pytest.raises(APIConnectionError):
+        embedder_config = EmbedderConfig(
+            embedding_provider="azure-openai",
+            embedding_api_key=pydantic.SecretStr("foo"),
+            embedding_azure_endpoint=f"https://localhost:{port}",
+        )
+        embedder = Embedder(config=embedder_config)
+        embedder.precheck()
+        _ = embedder.run(elements_filepath=embedder_file)
+
+    assert counter["POST"] == calls_before, (
+        f"Expected to see no change to POST calls toward embedder, got {counter} != {calls_before}"
+    )
+
+
+def test_openai_custom_tls_with_override_should_succeed(
+    mock_embeddings_server, monkeypatch, embedder_file: Path
+):
+    port, certificate_path, counter = mock_embeddings_server
+    calls_before = counter["POST"]
+    monkeypatch.setenv("REQUESTS_CA_BUNDLE", certificate_path)
+    embedder_config = EmbedderConfig(
+        embedding_provider="azure-openai",
+        embedding_api_key=pydantic.SecretStr("foo"),
+        embedding_azure_endpoint=f"https://localhost:{port}",
+    )
+    embedder = Embedder(config=embedder_config)
+    embedder.precheck()
+    results = embedder.run(elements_filepath=embedder_file)
+    assert results
+    assert counter["POST"] > calls_before, (
+        f"Expected to see more POST calls to embedder, got {counter} from {calls_before}"
+    )

unstructured_ingest/__version__.py

@@ -1 +1 @@
-__version__ = "1.0.59"  # pragma: no cover
+__version__ = "1.1.0"  # pragma: no cover

unstructured_ingest/embed/azure_openai.py

@@ -9,6 +9,7 @@
     OpenAIEmbeddingEncoder,
 )
 from unstructured_ingest.utils.dep_check import requires_dependencies
+from unstructured_ingest.utils.tls import ssl_context_with_optional_ca_override
 
 if TYPE_CHECKING:
     from openai import AsyncAzureOpenAI, AzureOpenAI
@@ -23,19 +24,23 @@ class AzureOpenAIEmbeddingConfig(OpenAIEmbeddingConfig):
 
     @requires_dependencies(["openai"], extras="openai")
     def get_client(self) -> "AzureOpenAI":
-        from openai import AzureOpenAI
+        from openai import AzureOpenAI, DefaultHttpxClient
 
+        client = DefaultHttpxClient(verify=ssl_context_with_optional_ca_override())
         return AzureOpenAI(
+            http_client=client,
             api_key=self.api_key.get_secret_value(),
             api_version=self.api_version,
             azure_endpoint=self.azure_endpoint,
         )
 
     @requires_dependencies(["openai"], extras="openai")
     def get_async_client(self) -> "AsyncAzureOpenAI":
-        from openai import AsyncAzureOpenAI
+        from openai import AsyncAzureOpenAI, DefaultAsyncHttpxClient
 
+        client = DefaultAsyncHttpxClient(verify=ssl_context_with_optional_ca_override())
         return AsyncAzureOpenAI(
+            http_client=client,
             api_key=self.api_key.get_secret_value(),
             api_version=self.api_version,
             azure_endpoint=self.azure_endpoint,

unstructured_ingest/embed/openai.py

@@ -18,6 +18,7 @@
 )
 from unstructured_ingest.logger import logger
 from unstructured_ingest.utils.dep_check import requires_dependencies
+from unstructured_ingest.utils.tls import ssl_context_with_optional_ca_override
 
 if TYPE_CHECKING:
     from openai import AsyncOpenAI, OpenAI
@@ -86,15 +87,21 @@ def run_precheck(self) -> None:
 
     @requires_dependencies(["openai"], extras="openai")
     def get_client(self) -> "OpenAI":
-        from openai import OpenAI
+        from openai import DefaultHttpxClient, OpenAI
 
-        return OpenAI(api_key=self.api_key.get_secret_value(), base_url=self.base_url)
+        client = DefaultHttpxClient(verify=ssl_context_with_optional_ca_override())
+        return OpenAI(
+            api_key=self.api_key.get_secret_value(), http_client=client, base_url=self.base_url
+        )
 
     @requires_dependencies(["openai"], extras="openai")
     def get_async_client(self) -> "AsyncOpenAI":
-        from openai import AsyncOpenAI
+        from openai import AsyncOpenAI, DefaultAsyncHttpxClient
 
-        return AsyncOpenAI(api_key=self.api_key.get_secret_value(), base_url=self.base_url)
+        client = DefaultAsyncHttpxClient(verify=ssl_context_with_optional_ca_override())
+        return AsyncOpenAI(
+            api_key=self.api_key.get_secret_value(), http_client=client, base_url=self.base_url
+        )
 
 
 @dataclass

unstructured_ingest/utils/tls.py

@@ -0,0 +1,15 @@
+import os
+import ssl
+
+import certifi
+
+
+def ssl_context_with_optional_ca_override():
+    """
+    # https://www.python-httpx.org/advanced/ssl/#working-with-ssl_cert_file-and-ssl_cert_dir
+    # We choose REQUESTS_CA_BUNDLE because that works with many other Python packages.
+    """
+    return ssl.create_default_context(
+        cafile=os.environ.get("REQUESTS_CA_BUNDLE", certifi.where()),
+        capath=os.environ.get("REQUESTS_CA_BUNDLE"),
+    )