feat(client-python): accept gateway cert by default via TOFU pre-fetch

clients/python/src/zb_mom_ww_mxgateway/options.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import ssl
 from collections.abc import Sequence
 from dataclasses import dataclass, field
 from pathlib import Path
@@ -9,6 +10,7 @@ from pathlib import Path
 import grpc
 
 from .auth import REDACTED, ApiKey
+from .errors import MxGatewayTransportError
 
 
 @dataclass(frozen=True)
@@ -19,6 +21,7 @@ class ClientOptions:
     api_key: str | ApiKey | None = None
     plaintext: bool = False
     ca_file: str | None = None
+    require_certificate_validation: bool = False
     server_name_override: str | None = None
     call_timeout: float | None = 30.0
     stream_timeout: float | None = None
@@ -45,6 +48,7 @@ class ClientOptions:
             f"{type(self).__name__}(endpoint={self.endpoint!r}, "
             f"api_key={api_key!r}, plaintext={self.plaintext!r}, "
             f"ca_file={self.ca_file!r}, "
+            f"require_certificate_validation={self.require_certificate_validation!r}, "
             f"server_name_override={self.server_name_override!r}, "
             f"call_timeout={self.call_timeout!r}, "
             f"stream_timeout={self.stream_timeout!r}, "
@@ -69,8 +73,22 @@ class BrowseChildrenOptions:
     historized_only: bool = False
 
 
+def _split_authority(endpoint: str) -> tuple[str, int]:
+    """Split a gRPC target (optionally scheme-prefixed) into (host, port)."""
+    target = endpoint.split("://", 1)[-1]
+    host, _, port = target.rpartition(":")
+    return (host or "localhost", int(port) if port else 443)
+
+
 def create_channel(options: ClientOptions) -> grpc.aio.Channel:
-    """Create a plaintext or TLS `grpc.aio` channel from client options."""
+    """Create a plaintext or TLS `grpc.aio` channel from client options.
+
+    The TLS default is lenient: grpc-python has no per-channel skip-verify, so
+    the server's presented certificate is fetched once (unverified) and pinned
+    as the channel's only trust root (trust-on-first-use). Set
+    `require_certificate_validation=True` to force system-trust verification, or
+    pass `ca_file` to verify against a specific CA — both bypass the TOFU path.
+    """
 
     channel_options: list[tuple[str, str | int]] = [
         ("grpc.max_receive_message_length", options.max_grpc_message_bytes),
@@ -82,11 +100,28 @@ def create_channel(options: ClientOptions) -> grpc.aio.Channel:
     if options.plaintext:
         return grpc.aio.insecure_channel(options.endpoint, options=channel_options)
 
-    root_certificates = None
     if options.ca_file:
         root_certificates = Path(options.ca_file).read_bytes()
+        credentials = grpc.ssl_channel_credentials(root_certificates=root_certificates)
+    elif options.require_certificate_validation:
+        credentials = grpc.ssl_channel_credentials()
+    else:
+        # Lenient default: grpc-python has no per-channel skip-verify, so fetch the
+        # server's certificate (unverified) and pin it for this channel (TOFU).
+        host, port = _split_authority(options.endpoint)
+        try:
+            presented = ssl.get_server_certificate((host, port))
+        except OSError as error:
+            raise MxGatewayTransportError(
+                f"failed to fetch TLS certificate from {options.endpoint}: {error}"
+            ) from error
+        credentials = grpc.ssl_channel_credentials(root_certificates=presented.encode("ascii"))
+        # The gateway self-signed cert always carries a "localhost" SAN, so default
+        # the SNI/target-name override to it when none was supplied, tolerating
+        # dial-by-IP or hostname mismatch.
+        if not options.server_name_override:
+            channel_options.append(("grpc.ssl_target_name_override", "localhost"))
 
-    credentials = grpc.ssl_channel_credentials(root_certificates=root_certificates)
     return grpc.aio.secure_channel(
         options.endpoint,
         credentials,