feat: add TransportOptions for configuring TLS, proxy, and default headers (#103)

Author: mridang

README.md

@@ -196,6 +196,84 @@ 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 provides a `TransportOptions` object that allows you to customise
+the underlying HTTP transport used for both OpenID discovery and API calls.
+
+### Disabling TLS Verification
+
+In development or testing environments with self-signed certificates, you can
+disable TLS verification entirely:
+
+```python
+from zitadel_client import Zitadel, TransportOptions
+
+options = TransportOptions(insecure=True)
+
+zitadel = Zitadel.with_client_credentials(
+    "https://your-instance.zitadel.cloud",
+    "client-id",
+    "client-secret",
+    transport_options=options,
+)
+```
+
+### Using a Custom CA Certificate
+
+If your Zitadel instance uses a certificate signed by a private CA, you can
+provide the path to the CA certificate in PEM format:
+
+```python
+from zitadel_client import Zitadel, TransportOptions
+
+options = TransportOptions(ca_cert_path="/path/to/ca.pem")
+
+zitadel = Zitadel.with_client_credentials(
+    "https://your-instance.zitadel.cloud",
+    "client-id",
+    "client-secret",
+    transport_options=options,
+)
+```
+
+### Custom Default Headers
+
+You can attach default headers to every outgoing request. This is useful for
+custom routing or tracing headers:
+
+```python
+from zitadel_client import Zitadel, TransportOptions
+
+options = TransportOptions(default_headers={"X-Custom-Header": "my-value"})
+
+zitadel = Zitadel.with_client_credentials(
+    "https://your-instance.zitadel.cloud",
+    "client-id",
+    "client-secret",
+    transport_options=options,
+)
+```
+
+### Proxy Configuration
+
+If your environment requires routing traffic through an HTTP proxy, you can
+specify the proxy URL. To authenticate with the proxy, embed the credentials
+directly in the URL:
+
+```python
+from zitadel_client import Zitadel, TransportOptions
+
+options = TransportOptions(proxy_url="http://user:pass@proxy:8080")
+
+zitadel = Zitadel.with_client_credentials(
+    "https://your-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

pyproject.toml

@@ -34,7 +34,7 @@ dev = [
   "pytest-cov>=2.8.1",
   "tox>=3.9.0",
   "types-python-dateutil>=2.8.19.14",
-  "testcontainers==3.7.1",
+  "testcontainers>=4.14.0,<5.0.0",
   "python-dotenv==1.1.1",
   "ruff>=0.12.4",
   "sphinx==7.4.7",

test/fixtures/ca.pem

@@ -0,0 +1,20 @@
+-----BEGIN CERTIFICATE-----
+MIIDOjCCAiKgAwIBAgIUYtCHt3J95fUpagYaFNw8M1/oV7kwDQYJKoZIhvcNAQEL
+BQAwFDESMBAGA1UEAwwJbG9jYWxob3N0MCAXDTI2MDMwNDA1MTcwNVoYDzIxMjYw
+MjA4MDUxNzA1WjAUMRIwEAYDVQQDDAlsb2NhbGhvc3QwggEiMA0GCSqGSIb3DQEB
+AQUAA4IBDwAwggEKAoIBAQCVY1jORnqyVB9tUgYYo9U3uYCVtCzWt3lGCoxDpxAb
+LlpNnqOxG33ugRbNTY/QBht37Q37PjBahMJxkRE7EPsqi2Bz2fsZMyB7pJgP5iTA
+0cILFyFzGpgUkXjmtsozKy0jAHpnzHGALjtzoKgp4SxCrWSp/MYtfMkBP9xbEpf1
+IYYQyyiISgic0/vO+nUEjyR/ULFP+nd48KjOHwWIHqwMY3nuzqScshAsyIZzSRT0
+ND2TLK1rxGoITqsOg2yTxRWwP0khvE08Y/59BGfWZq0svBCp2E3sIXg2Z3hlie7o
+n+3P0F00kQfrEvkTi/cHv2vuhJpnlHxmTgJBRwhWE2+xAgMBAAGjgYEwfzAdBgNV
+HQ4EFgQUPOzmGXHMRu3zIZqKLad8EkHkZvowHwYDVR0jBBgwFoAUPOzmGXHMRu3z
+IZqKLad8EkHkZvowDwYDVR0TAQH/BAUwAwEB/zAsBgNVHREEJTAjgglsb2NhbGhv
+c3SHBH8AAAGHEAAAAAAAAAAAAAAAAAAAAAEwDQYJKoZIhvcNAQELBQADggEBAD/z
+IRzYSBp6qPrvVgIX5/mEwN6ylp1J1pTC8nPQRozg0X2SEnRxz1DGBa1l046QVew2
+3+LuGYWtVkTzEtiX7BN2jSshX8d8Ss73+psZOye6t8VcAmEeVVdnqU+EzVAhM1DP
+mUiNxJPHgK2cZkpV2BHB0Ccu7qVfaIFvTk2OdbGOsQ7+r2l562kUDzCFvBo/mskO
+xiIt3YMZrpyLJJzvgi+fIo351oqLvTKOHw30FelAPIHo/A2OgngsM31HvwxROYlr
+C5mET6wnOtjTQbKORADTGQ8D3sJCjQJ/AI34Q4C2q/PBljVL8JKoAPzwviYAuqdd
+NIIKpaYUzng24gw7+50=
+-----END CERTIFICATE-----

test/fixtures/keystore.p12

@@ -0,0 +1,19 @@
+{
+  "request": {
+    "method": "GET",
+    "url": "/.well-known/openid-configuration"
+  },
+  "response": {
+    "status": 200,
+    "headers": {
+      "Content-Type": "application/json"
+    },
+    "jsonBody": {
+      "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"
+    }
+  }
+}

test/fixtures/mappings/oidc-discovery.json

@@ -0,0 +1,16 @@
+{
+  "request": {
+    "method": "POST",
+    "url": "/zitadel.settings.v2.SettingsService/GetGeneralSettings"
+  },
+  "response": {
+    "status": 200,
+    "headers": {
+      "Content-Type": "application/json"
+    },
+    "jsonBody": {
+      "defaultLanguage": "{{request.scheme}}",
+      "defaultOrgId": "{{request.headers.X-Custom-Header}}"
+    }
+  }
+}

test/fixtures/mappings/settings.json

@@ -0,0 +1,17 @@
+{
+  "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
+    }
+  }
+}

test/fixtures/mappings/token.json

@@ -0,0 +1,3 @@
+http_port 3128
+acl all src all
+http_access allow all

test/fixtures/squid.conf

@@ -0,0 +1,39 @@
+import unittest
+
+from zitadel_client.transport_options import TransportOptions
+
+
+class TransportOptionsTest(unittest.TestCase):
+    def test_defaults_returns_empty(self) -> None:
+        self.assertEqual({}, TransportOptions.defaults().to_session_kwargs())
+
+    def test_insecure_sets_verify_false(self) -> None:
+        opts = TransportOptions(insecure=True)
+        self.assertEqual({"verify": False}, opts.to_session_kwargs())
+
+    def test_ca_cert_path_sets_verify(self) -> None:
+        opts = TransportOptions(ca_cert_path="/path/to/ca.pem")
+        self.assertEqual({"verify": "/path/to/ca.pem"}, opts.to_session_kwargs())
+
+    def test_proxy_url_sets_proxies(self) -> None:
+        opts = TransportOptions(proxy_url="http://proxy:3128")
+        self.assertEqual(
+            {"proxies": {"http": "http://proxy:3128", "https": "http://proxy:3128"}},
+            opts.to_session_kwargs(),
+        )
+
+    def test_insecure_takes_precedence_over_ca_cert(self) -> None:
+        opts = TransportOptions(insecure=True, ca_cert_path="/nonexistent/ca.pem")
+        self.assertEqual({"verify": False}, opts.to_session_kwargs())
+
+    def test_immutability(self) -> None:
+        opts = TransportOptions.defaults()
+        with self.assertRaises(AttributeError):
+            opts.insecure = True  # type: ignore[misc]
+
+    def test_defaults_factory(self) -> None:
+        opts = TransportOptions.defaults()
+        self.assertEqual({}, dict(opts.default_headers))
+        self.assertIsNone(opts.ca_cert_path)
+        self.assertFalse(opts.insecure)
+        self.assertIsNone(opts.proxy_url)

test/test_transport_options.py

@@ -1,17 +1,32 @@
 import importlib
 import inspect
+import os
 import pkgutil
 import unittest
+import urllib.request
+from typing import Optional
+
+from testcontainers.core.container import DockerContainer
+from testcontainers.core.network import Network
+from testcontainers.core.wait_strategies import PortWaitStrategy
+from testcontainers.core.waiting_utils import wait_container_is_ready
 
 from zitadel_client.auth.no_auth_authenticator import NoAuthAuthenticator
+from zitadel_client.transport_options import TransportOptions
 from zitadel_client.zitadel import Zitadel
 
+FIXTURES_DIR = os.path.join(os.path.dirname(__file__), "fixtures")
+
+
+@wait_container_is_ready()
+def _wait_for_wiremock(host: str, port: str) -> None:
+    url = f"http://{host}:{port}/__admin/mappings"
+    with urllib.request.urlopen(url, timeout=5) as resp:  # noqa: S310
+        if resp.status != 200:
+            raise ConnectionError(f"WireMock not ready: {resp.status}")
+
 
 class ZitadelServicesTest(unittest.TestCase):
-    """
-    Test to verify that all API service classes defined in the "zitadel_client.api" namespace
-    are registered as attributes in the Zitadel class.
-    """
 
     def test_services_dynamic(self) -> None:
         expected = set()
@@ -30,3 +45,114 @@ def test_services_dynamic(self) -> None:
             and getattr(zitadel, attr).__class__.__module__.startswith("zitadel_client.api")
         }
         self.assertEqual(expected, actual)
+
+
+class ZitadelTransportTest(unittest.TestCase):
+    host: Optional[str] = None
+    http_port: Optional[str] = None
+    https_port: Optional[str] = None
+    proxy_port: Optional[str] = None
+    ca_cert_path: Optional[str] = None
+    wiremock: DockerContainer = None
+    proxy: DockerContainer = None
+    network: Network = None
+
+    @classmethod
+    def setup_class(cls) -> None:
+        cls.ca_cert_path = os.path.join(FIXTURES_DIR, "ca.pem")
+        keystore_path = os.path.join(FIXTURES_DIR, "keystore.p12")
+        squid_conf = os.path.join(FIXTURES_DIR, "squid.conf")
+
+        cls.network = Network().create()
+
+        cls.wiremock = (
+            DockerContainer("wiremock/wiremock:3.12.1")
+            .with_network(cls.network)
+            .with_network_aliases("wiremock")
+            .with_exposed_ports(8080, 8443)
+            .with_volume_mapping(keystore_path, "/home/wiremock/keystore.p12", mode="ro")
+            .with_volume_mapping(
+                os.path.join(FIXTURES_DIR, "mappings"), "/home/wiremock/mappings", mode="ro"
+            )
+            .with_command(
+                "--https-port 8443"
+                " --https-keystore /home/wiremock/keystore.p12"
+                " --keystore-password password"
+                " --keystore-type PKCS12"
+                " --global-response-templating"
+            )
+        )
+        cls.wiremock.start()
+
+        cls.proxy = (
+            DockerContainer("ubuntu/squid:6.10-24.10_beta")
+            .with_network(cls.network)
+            .with_exposed_ports(3128)
+            .with_volume_mapping(squid_conf, "/etc/squid/squid.conf", mode="ro")
+            .waiting_for(PortWaitStrategy(3128))
+        )
+        cls.proxy.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)
+        cls.proxy_port = cls.proxy.get_exposed_port(3128)
+
+        _wait_for_wiremock(cls.host, cls.http_port)
+
+    @classmethod
+    def teardown_class(cls) -> None:
+        if cls.proxy is not None:
+            cls.proxy.stop()
+        if cls.wiremock is not None:
+            cls.wiremock.stop()
+        if cls.network is not None:
+            cls.network.remove()
+
+    def test_custom_ca_cert(self) -> None:
+        zitadel = Zitadel.with_client_credentials(
+            f"https://{self.host}:{self.https_port}",
+            "dummy-client",
+            "dummy-secret",
+            transport_options=TransportOptions(ca_cert_path=self.ca_cert_path),
+        )
+        response = zitadel.settings.get_general_settings({})
+        self.assertEqual("https", response.default_language)
+
+    def test_insecure_mode(self) -> None:
+        zitadel = Zitadel.with_client_credentials(
+            f"https://{self.host}:{self.https_port}",
+            "dummy-client",
+            "dummy-secret",
+            transport_options=TransportOptions(insecure=True),
+        )
+        response = zitadel.settings.get_general_settings({})
+        self.assertEqual("https", response.default_language)
+
+    def test_default_headers(self) -> None:
+        zitadel = Zitadel.with_client_credentials(
+            f"http://{self.host}:{self.http_port}",
+            "dummy-client",
+            "dummy-secret",
+            transport_options=TransportOptions(default_headers={"X-Custom-Header": "test-value"}),
+        )
+        response = zitadel.settings.get_general_settings({})
+        self.assertEqual("http", response.default_language)
+        self.assertEqual("test-value", response.default_org_id)
+
+    def test_proxy_url(self) -> None:
+        zitadel = Zitadel.with_access_token(
+            "http://wiremock:8080",
+            "test-token",
+            transport_options=TransportOptions(proxy_url=f"http://{self.host}:{self.proxy_port}"),
+        )
+        response = zitadel.settings.get_general_settings({})
+        self.assertEqual("http", response.default_language)
+
+    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",
+            )