Add TransportOptions for configuring TLS, proxy, and default headers

Introduce a TransportOptions dataclass that encapsulates transport-level
configuration (default_headers, ca_cert_path, insecure, proxy_url). Factory
methods now accept an optional transport_options parameter alongside the
existing individual parameters for backward compatibility. Internal auth
plumbing (OpenId, builders) refactored to use TransportOptions throughout.

Author: mridang

README.md

@@ -196,6 +196,97 @@ Choose the authentication method that best suits your needs based on your
 environment and security requirements. For more details, please refer to the
 [Zitadel documentation on authenticating service users](https://zitadel.com/docs/guides/integrate/service-users/authenticate-service-users).
 
+## Advanced Configuration
+
+The SDK factory methods (`with_client_credentials`, `with_private_key`,
+`with_access_token`) accept additional keyword arguments for advanced
+transport configuration.
+
+### Disabling TLS Verification
+
+To disable TLS certificate verification (useful for development with
+self-signed certificates), pass `insecure=True`:
+
+```python
+import zitadel_client as zitadel
+
+client = zitadel.Zitadel.with_client_credentials(
+    "https://example.us1.zitadel.cloud",
+    "client-id",
+    "client-secret",
+    insecure=True,
+)
+```
+
+### Using a Custom CA Certificate
+
+To use a custom CA certificate for TLS verification, pass
+`ca_cert_path` with the path to your CA certificate file:
+
+```python
+import zitadel_client as zitadel
+
+client = zitadel.Zitadel.with_client_credentials(
+    "https://example.us1.zitadel.cloud",
+    "client-id",
+    "client-secret",
+    ca_cert_path="/path/to/ca.pem",
+)
+```
+
+### Custom Default Headers
+
+To send custom headers with every request, pass a `default_headers`
+dictionary:
+
+```python
+import zitadel_client as zitadel
+
+client = zitadel.Zitadel.with_client_credentials(
+    "https://example.us1.zitadel.cloud",
+    "client-id",
+    "client-secret",
+    default_headers={"Proxy-Authorization": "Basic ..."},
+)
+```
+
+### Proxy Configuration
+
+To route all SDK traffic through an HTTP proxy, pass a `proxy_url`:
+
+```python
+import zitadel_client as zitadel
+
+client = zitadel.Zitadel.with_client_credentials(
+    "https://example.us1.zitadel.cloud",
+    "client-id",
+    "client-secret",
+    proxy_url="http://proxy:8080",
+)
+```
+
+### Using TransportOptions
+
+All transport settings can be combined into a single `TransportOptions` object:
+
+```python
+from zitadel_client import Zitadel, TransportOptions
+
+options = TransportOptions(
+    insecure=True,
+    ca_cert_path="/path/to/ca.pem",
+    default_headers={"Proxy-Authorization": "Basic dXNlcjpwYXNz"},
+    proxy_url="http://proxy:8080",
+)
+
+zitadel = Zitadel.with_client_credentials(
+    "https://my-instance.zitadel.cloud",
+    "client-id",
+    "client-secret",
+    transport_options=options,
+)
+```
+
 ## Design and Dependencies
 
 This SDK is designed to be lean and efficient, focusing on providing a

test/test_transport_options.py

@@ -0,0 +1,190 @@
+import json
+import os
+import ssl
+import tempfile
+import time
+import unittest
+import urllib.request
+from typing import Optional
+
+from testcontainers.core.container import DockerContainer
+
+from zitadel_client.transport_options import TransportOptions
+from zitadel_client.zitadel import Zitadel
+
+
+class TransportOptionsTest(unittest.TestCase):
+    """
+    Test class for verifying transport options (default_headers, ca_cert_path, insecure)
+    on the Zitadel factory methods.
+
+    This class starts a Docker container running WireMock with HTTPS support before any
+    tests run and stops it after all tests. It registers stubs for OpenID Configuration
+    discovery and the token endpoint so that Zitadel.with_client_credentials() can
+    complete its initialization flow.
+    """
+
+    host: Optional[str] = None
+    http_port: Optional[str] = None
+    https_port: Optional[str] = None
+    ca_cert_path: Optional[str] = None
+    wiremock: DockerContainer = None
+
+    @classmethod
+    def setup_class(cls) -> None:
+        cls.wiremock = (
+            DockerContainer("wiremock/wiremock:3.3.1")
+            .with_exposed_ports(8080, 8443)
+            .with_command("--https-port 8443 --global-response-templating")
+        )
+        cls.wiremock.start()
+
+        cls.host = cls.wiremock.get_container_host_ip()
+        cls.http_port = cls.wiremock.get_exposed_port(8080)
+        cls.https_port = cls.wiremock.get_exposed_port(8443)
+
+        # Wait for WireMock to be ready by polling the admin API
+        admin_url = f"http://{cls.host}:{cls.http_port}/__admin/mappings"
+        for _ in range(30):
+            try:
+                with urllib.request.urlopen(admin_url, timeout=2) as resp:  # noqa: S310
+                    if resp.status == 200:
+                        break
+            except Exception:  # noqa: S110, BLE001
+                pass
+            time.sleep(1)
+
+        # Register stub for OpenID Configuration discovery
+        oidc_stub = json.dumps(
+            {
+                "request": {"method": "GET", "url": "/.well-known/openid-configuration"},
+                "response": {
+                    "status": 200,
+                    "headers": {"Content-Type": "application/json"},
+                    "body": (
+                        '{"issuer":"{{request.baseUrl}}",'
+                        '"token_endpoint":"{{request.baseUrl}}/oauth/v2/token",'
+                        '"authorization_endpoint":"{{request.baseUrl}}/oauth/v2/authorize",'
+                        '"userinfo_endpoint":"{{request.baseUrl}}/oidc/v1/userinfo",'
+                        '"jwks_uri":"{{request.baseUrl}}/oauth/v2/keys"}'
+                    ),
+                },
+            }
+        ).encode()
+
+        req = urllib.request.Request(
+            f"http://{cls.host}:{cls.http_port}/__admin/mappings",
+            data=oidc_stub,
+            headers={"Content-Type": "application/json"},
+            method="POST",
+        )
+        with urllib.request.urlopen(req) as resp:  # noqa: S310
+            assert resp.status == 201
+
+        # Register stub for the token endpoint
+        token_stub = json.dumps(
+            {
+                "request": {"method": "POST", "url": "/oauth/v2/token"},
+                "response": {
+                    "status": 200,
+                    "headers": {"Content-Type": "application/json"},
+                    "jsonBody": {
+                        "access_token": "test-token-12345",
+                        "token_type": "Bearer",
+                        "expires_in": 3600,
+                    },
+                },
+            }
+        ).encode()
+
+        req = urllib.request.Request(
+            f"http://{cls.host}:{cls.http_port}/__admin/mappings",
+            data=token_stub,
+            headers={"Content-Type": "application/json"},
+            method="POST",
+        )
+        with urllib.request.urlopen(req) as resp:  # noqa: S310
+            assert resp.status == 201
+
+        # Extract the WireMock HTTPS certificate to a temp file
+        pem_cert = ssl.get_server_certificate((cls.host, int(cls.https_port)))
+        cert_file = tempfile.NamedTemporaryFile(suffix=".pem", delete=False)
+        cert_file.write(pem_cert.encode())
+        cert_file.close()
+        cls.ca_cert_path = cert_file.name
+
+    @classmethod
+    def teardown_class(cls) -> None:
+        if cls.ca_cert_path is not None:
+            os.unlink(cls.ca_cert_path)
+        if cls.wiremock is not None:
+            cls.wiremock.stop()
+
+    def test_custom_ca_cert(self) -> None:
+        zitadel = Zitadel.with_client_credentials(
+            f"https://{self.host}:{self.https_port}",
+            "dummy-client",
+            "dummy-secret",
+            ca_cert_path=self.ca_cert_path,
+        )
+        self.assertIsNotNone(zitadel)
+
+    def test_insecure_mode(self) -> None:
+        zitadel = Zitadel.with_client_credentials(
+            f"https://{self.host}:{self.https_port}",
+            "dummy-client",
+            "dummy-secret",
+            insecure=True,
+        )
+        self.assertIsNotNone(zitadel)
+
+    def test_default_headers(self) -> None:
+        # Use HTTP to avoid TLS concerns
+        zitadel = Zitadel.with_client_credentials(
+            f"http://{self.host}:{self.http_port}",
+            "dummy-client",
+            "dummy-secret",
+            default_headers={"X-Custom-Header": "test-value"},
+        )
+        self.assertIsNotNone(zitadel)
+
+        # Verify via WireMock request journal
+        journal_url = f"http://{self.host}:{self.http_port}/__admin/requests"
+        with urllib.request.urlopen(journal_url) as response:  # noqa: S310
+            journal = json.loads(response.read().decode())
+
+        found_header = False
+        for req in journal.get("requests", []):
+            headers = req.get("request", {}).get("headers", {})
+            if "X-Custom-Header" in headers:
+                found_header = True
+                break
+        self.assertTrue(found_header, "Custom header should be present in WireMock request journal")
+
+    def test_proxy_url(self) -> None:
+        # Use HTTP (not HTTPS) to avoid TLS complications with the proxy
+        zitadel = Zitadel.with_client_credentials(
+            f"http://{self.host}:{self.http_port}",
+            "dummy-client",
+            "dummy-secret",
+            proxy_url=f"http://{self.host}:{self.http_port}",
+        )
+        self.assertIsNotNone(zitadel)
+
+    def test_no_ca_cert_fails(self) -> None:
+        with self.assertRaises(Exception):  # noqa: B017
+            Zitadel.with_client_credentials(
+                f"https://{self.host}:{self.https_port}",
+                "dummy-client",
+                "dummy-secret",
+            )
+
+    def test_transport_options_object(self) -> None:
+        opts = TransportOptions(insecure=True)
+        zitadel = Zitadel.with_client_credentials(
+            f"https://{self.host}:{self.https_port}",
+            "dummy-client",
+            "dummy-secret",
+            transport_options=opts,
+        )
+        self.assertIsNotNone(zitadel)

uv.lock

@@ -8,4 +8,5 @@
     ZitadelError,  # noqa F401
 )
 from .models import *  # noqa: F403, F401
+from .transport_options import TransportOptions  # noqa F401
 from .zitadel import Zitadel  # noqa F401

zitadel_client/__init__.py

@@ -65,6 +65,8 @@ def __init__(
 
         self.rest_client = rest.RESTClientObject(configuration)
         self.default_headers = {"User-Agent": configuration.user_agent}
+        if configuration.default_headers:
+            self.default_headers.update(configuration.default_headers)
         if header_name is not None and header_value is not None:
             self.default_headers[header_name] = header_value
         self.client_side_validation = configuration.client_side_validation

zitadel_client/api_client.py

@@ -1,5 +1,5 @@
 import sys
-from typing import Dict, Set
+from typing import Dict, Optional, Set
 
 if sys.version_info >= (3, 12):
     from typing import override
@@ -13,6 +13,7 @@
     OAuthAuthenticatorBuilder,
 )
 from zitadel_client.auth.open_id import OpenId
+from zitadel_client.transport_options import TransportOptions
 
 
 class ClientCredentialsAuthenticator(OAuthAuthenticator):
@@ -50,16 +51,19 @@ def get_grant(self) -> Dict[str, str]:
         return {"grant_type": "client_credentials"}
 
     @staticmethod
-    def builder(host: str, client_id: str, client_secret: str) -> "ClientCredentialsAuthenticatorBuilder":
+    def builder(
+        host: str, client_id: str, client_secret: str, transport_options: Optional[TransportOptions] = None
+    ) -> "ClientCredentialsAuthenticatorBuilder":
         """
         Returns a builder for constructing a ClientCredentialsAuthenticator.
 
         :param host: The base URL for the OAuth provider.
         :param client_id: The OAuth client identifier.
         :param client_secret: The OAuth client secret.
+        :param transport_options: Optional TransportOptions for configuring HTTP connections.
         :return: A ClientCredentialsAuthenticatorBuilder instance.
         """
-        return ClientCredentialsAuthenticatorBuilder(host, client_id, client_secret)
+        return ClientCredentialsAuthenticatorBuilder(host, client_id, client_secret, transport_options=transport_options)
 
 
 class ClientCredentialsAuthenticatorBuilder(OAuthAuthenticatorBuilder["ClientCredentialsAuthenticatorBuilder"]):
@@ -70,15 +74,16 @@ class ClientCredentialsAuthenticatorBuilder(OAuthAuthenticatorBuilder["ClientCre
     required for the client credentials flow.
     """
 
-    def __init__(self, host: str, client_id: str, client_secret: str):
+    def __init__(self, host: str, client_id: str, client_secret: str, transport_options: Optional[TransportOptions] = None):
         """
         Initializes the ClientCredentialsAuthenticatorBuilder with host, client ID, and client secret.
 
         :param host: The base URL for the OAuth provider.
         :param client_id: The OAuth client identifier.
         :param client_secret: The OAuth client secret.
+        :param transport_options: Optional TransportOptions for configuring HTTP connections.
         """
-        super().__init__(host)
+        super().__init__(host, transport_options=transport_options)
         self.client_id = client_id
         self.client_secret = client_secret