feat: add TransportOptions for configuring TLS, proxy, and default headers

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/squid.conf

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

test/test_transport_options.py

@@ -0,0 +1,218 @@
+import json
+import os
+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.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 TransportOptionsTest(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.3.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_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)
+
+        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
+
+        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
+
+        settings_stub = json.dumps(
+            {
+                "request": {
+                    "method": "POST",
+                    "url": "/zitadel.settings.v2.SettingsService/GetGeneralSettings",
+                },
+                "response": {
+                    "status": 200,
+                    "headers": {"Content-Type": "application/json"},
+                    "jsonBody": {},
+                },
+            }
+        ).encode()
+
+        req = urllib.request.Request(
+            f"http://{cls.host}:{cls.http_port}/__admin/mappings",
+            data=settings_stub,
+            headers={"Content-Type": "application/json"},
+            method="POST",
+        )
+        with urllib.request.urlopen(req) as resp:  # noqa: S310
+            assert resp.status == 201
+
+    @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),
+        )
+        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",
+            transport_options=TransportOptions(insecure=True),
+        )
+        self.assertIsNotNone(zitadel)
+
+    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"}),
+        )
+        self.assertIsNotNone(zitadel)
+
+        zitadel.settings.get_general_settings({})
+
+        verify_body = json.dumps(
+            {
+                "url": "/zitadel.settings.v2.SettingsService/GetGeneralSettings",
+                "headers": {"X-Custom-Header": {"equalTo": "test-value"}},
+            }
+        ).encode()
+        req = urllib.request.Request(
+            f"http://{self.host}:{self.http_port}/__admin/requests/count",
+            data=verify_body,
+            headers={"Content-Type": "application/json"},
+            method="POST",
+        )
+        with urllib.request.urlopen(req) as resp:  # noqa: S310
+            result = json.loads(resp.read().decode())
+        self.assertGreaterEqual(result["count"], 1, "Custom header should be present on API call")
+
+    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}"),
+        )
+        self.assertIsNotNone(zitadel)
+        zitadel.settings.get_general_settings({})
+
+    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",
+            )

zitadel_client/__init__.py

@@ -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