Allow explicit verified TLS contexts for private certificate authorities

.github/workflows/ci.yml

@@ -158,6 +158,7 @@ jobs:
         shell: bash
         run: |
           set -o pipefail
+          command -v openssl
           # Leave room within the 15-minute job for bounded failure diagnostics.
           timeout --signal=INT --kill-after=10s 13m \
             pytest tests/ -m "not integration" -vv \

CHANGELOG.md

@@ -11,6 +11,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Nexus operation mocks in `WorkflowEnvironment`, keyed by endpoint, service
   and operation. Register canned responses or synchronous callables; workflow
   code receives completion, acceptance or `NexusOperationFailed` through replay.
+- Explicit verified `tls_context` configuration for async and synchronous
+  clients, including cloned query clients and runtime payload transfers.
+  Applications can add a private CA while retaining certifi's public roots.
 
 ### Fixed
 

CONTRIBUTING.md

@@ -2,6 +2,10 @@
 
 Run Ruff, mypy, and focused pytest coverage for changed code.
 
+The TLS tests use the OpenSSL CLI to create a throwaway private CA and run a
+local HTTPS endpoint. Install `openssl` to run those controls locally; the
+supported-version CI requires it before running tests.
+
 Replay and payload-codec fixes also follow the organization
 [regression-corpus contract](https://github.com/durable-workflow/.github/tree/main/regression-corpus).
 Put minimal replay histories in the existing golden-history format when

docs/reference/client.md

@@ -1,5 +1,48 @@
 # Client
 
+## Trust a private certificate authority
+
+The `tls_context` option is available in unreleased SDK source and will be
+included in the next package release.
+
+Pass a verifying `ssl.SSLContext` to configure trust for this client without
+changing trust for other clients in the process. Start with certifi's public
+roots, then add your private CA:
+
+```python
+import ssl
+
+import certifi
+
+from durable_workflow import Client
+
+context = ssl.create_default_context(cafile=certifi.where())
+context.load_verify_locations(cafile="company-ca.pem")
+
+client = Client("https://workflow.example.internal", tls_context=context)
+```
+
+The synchronous `durable_workflow.sync.Client` accepts the same option. Worker
+query clients and Nexus namespace views reuse the context. Control and worker
+requests, including runtime-owned external payload upload and fetch, use the
+same configured transport. Separately configured storage drivers own their
+storage-service transports.
+
+An explicit context replaces httpx's default root selection. It does not
+automatically add public roots: the example above keeps them by loading
+certifi before adding the private CA. Contexts with disabled certificate or
+hostname verification are rejected. Treat the supplied context as read-only
+after client construction.
+
+With no explicit context, the existing httpx behavior remains: trust comes
+from the installed certifi bundle, or from `SSL_CERT_FILE` / `SSL_CERT_DIR`
+when those environment variables are set. Installing a CA only into the OS
+store does not update certifi. `SSL_CERT_FILE` replaces the root bundle for
+httpx clients throughout the process; a private-CA-only file can therefore
+remove public trust. Prefer a per-client context when applications need
+different trust configurations. An explicit context takes precedence over
+those environment root settings.
+
 `Client.execute_nexus_operation(...)` calls a registered Nexus service
 operation through the service-catalog API and returns `NexusOperationResult`.
 Workflow authors should normally yield `ctx.call_nexus_service(...)`; the

docs/sdk-reference.md

@@ -966,6 +966,16 @@ compatibility manifest. A client configured with only the opposite role's token
 still fails before transport for actual worker or control operations; use
 `token` when one shared credential intentionally authorizes both planes.
 
+## TLS trust
+
+Unreleased SDK source accepts `tls_context=ssl_context` on async and synchronous
+clients. Use a verifying `ssl.SSLContext` initialized with certifi's public
+roots, then load your private CA. Worker query clones and runtime-owned payload
+transfers reuse that trust configuration. See the
+[private CA recipe](reference/client.md#trust-a-private-certificate-authority)
+for root selection and `SSL_CERT_FILE` / `SSL_CERT_DIR` behavior. Published
+packages gain this option with the next SDK release.
+
 ## Poll failures
 
 The backoff described here is available in unreleased SDK source. Published

src/durable_workflow/client.py

@@ -26,6 +26,7 @@
 import math
 import os
 import re
+import ssl
 import time
 import uuid
 import warnings
@@ -1631,6 +1632,10 @@ class Client:
 
     The client owns one `httpx.AsyncClient` connection pool. Use it as an async
     context manager or call `aclose()` when finished.
+
+    ``tls_context`` selects an explicit trust configuration. It must require
+    certificate and hostname verification. Treat the context as read-only after
+    construction; worker query clones and Nexus namespace views reuse it.
     """
 
     def __init__(
@@ -1642,6 +1647,7 @@ def __init__(
         worker_token: str | None = None,
         namespace: str = "default",
         timeout: float = 60.0,
+        tls_context: ssl.SSLContext | None = None,
         retry_policy: TransportRetryPolicy | None = None,
         metrics: MetricsRecorder | None = None,
         payload_size_limit_bytes: int = serializer.DEFAULT_PAYLOAD_SIZE_BYTES,
@@ -1652,6 +1658,11 @@ def __init__(
         external_storage_cache: ExternalPayloadCache | None = None,
         telemetry: ExecutionTelemetry | None = None,
     ) -> None:
+        if tls_context is not None:
+            if not isinstance(tls_context, ssl.SSLContext):
+                raise TypeError("tls_context must be an ssl.SSLContext")
+            if tls_context.verify_mode != ssl.CERT_REQUIRED or not tls_context.check_hostname:
+                raise ValueError("tls_context must enable certificate and hostname verification")
         self.base_url = _normalize_base_url(base_url)
         self.token = token
         self.control_token = control_token
@@ -1660,6 +1671,7 @@ def __init__(
         self.telemetry = telemetry
         self._observability_capability = False
         self.timeout = timeout
+        self._tls_context = tls_context
         self.retry_policy = retry_policy or TransportRetryPolicy()
         self.metrics = metrics or NOOP_METRICS
         self.payload_size_warning_config = (
@@ -1679,7 +1691,9 @@ def __init__(
             if external_storage_cache is not None
             else ExternalPayloadCache()
         )
-        self._http = httpx.AsyncClient(base_url=self.base_url, timeout=timeout)
+        self._http = httpx.AsyncClient(
+            base_url=self.base_url, timeout=timeout, verify=tls_context if tls_context is not None else True,
+        )
         self._cluster_info: dict[str, Any] | None = None
         self._cluster_info_lock = asyncio.Lock()
         self._runtime_external_payload_transport_cache: (
@@ -1688,6 +1702,11 @@ def __init__(
         self._runtime_external_payload_transport_resolved = False
         self._workflow_payload_observers: dict[str, Callable[[dict[str, Any]], Awaitable[None]]] = {}
 
+    @property
+    def tls_context(self) -> ssl.SSLContext | None:
+        """The explicit TLS context, or ``None`` for httpx's default trust selection."""
+        return self._tls_context
+
     async def aclose(self) -> None:
         """Close the underlying ``httpx`` connection pool.
 

src/durable_workflow/sync.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import asyncio
+import ssl
 from collections.abc import Coroutine
 from typing import Any, TypeVar
 
@@ -266,6 +267,9 @@ class Client:
 
     The client keeps one event loop for its lifetime so its async HTTP connection
     pool and returned handles always run on the loop that owns them.
+
+    ``tls_context`` forwards the same verified trust configuration accepted by
+    the async client. Treat the supplied context as read-only during use.
     """
 
     def __init__(
@@ -277,6 +281,7 @@ def __init__(
         worker_token: str | None = None,
         namespace: str = "default",
         timeout: float = 60.0,
+        tls_context: ssl.SSLContext | None = None,
         retry_policy: TransportRetryPolicy | None = None,
         metrics: MetricsRecorder | None = None,
         external_storage: ExternalStorageDriver | None = None,
@@ -290,6 +295,7 @@ def __init__(
             worker_token=worker_token,
             namespace=namespace,
             timeout=timeout,
+            tls_context=tls_context,
             retry_policy=retry_policy,
             metrics=metrics,
             external_storage=external_storage,

src/durable_workflow/worker.py

@@ -4110,6 +4110,7 @@ def _clone_client_for_query_tasks(self) -> Client:
             worker_token=self.client.worker_token,
             namespace=self.client.namespace,
             timeout=getattr(self.client, "timeout", 60.0),
+            tls_context=self.client.tls_context,
             retry_policy=self.client.retry_policy,
             metrics=self.metrics,
             payload_size_limit_bytes=(