feat(spanner): auth login support for Spanner Omni endpoints (#18273)

**⚠️ Note to Reviewers:** 
> The files under `google/cloud/spanner_v1/omni/proto/`
(`authentication_pb2*`, `login_pb2*`, `login_pb2_grpc*`) are
auto-generated.
> 
> The core hand-written logic is contained in the OPAQUE cryptography
module, `LoginClient`, `SpannerOmniCredentials`, and the client/DB-API
integrations.

This PR introduces native username/password authentication support for
Spanner Omni endpoints using the OPAQUE password-authenticated key
exchange (PAKE) protocol, maintaining parity with the Go and Java client
implementations.

Java implementation -
googleapis/google-cloud-java#13470
Go implementation -
googleapis/google-cloud-go#20085

**Key Changes:**
* **Omni Login Protocol:** Added generated protobufs
(`authentication_pb2*`, `login_pb2*`, `login_pb2_grpc*`) and a gRPC
`LoginClient` to handle the authentication handshake with Spanner Omni
endpoints over the `LoginService/Login` bi-directional stream.
* **OPAQUE Protocol (`opaque.py`):** Implements the client-side OPAQUE
protocol utilizing NIST P-256 elliptic curve arithmetic
(`cryptography`), Argon2id stretching with defensive parameter
validation, RFC 9380 hash-to-curve / random oracle mapping, and
constant-time MAC verification.
* **Security & Memory Safety:** Ensures in-place zeroization of
intermediate keys, stretched passwords, and shared secrets using mutable
byte buffers inside `finally` blocks.
* **SpannerOmniCredentials & Interceptors (`credentials.py`):**
Implements `SpannerOmniCredentials` subclassing
`google.auth.credentials.Credentials` to manage Bearer access token
lifecycle, automatic expiry checks, and transparent token refreshes.
Provides dedicated interceptors for both synchronous `grpc` and
asynchronous `grpc.aio` channels across all RPC patterns (unary-unary,
unary-stream, stream-unary, stream-stream).
* **Client & DB-API Integration:**
- Extended `spanner_v1.Client` and `spanner_v1.AsyncClient` to accept
`username`, `password`, and `instance_type="omni"`, automatically wiring
up `SpannerOmniCredentials`.
- Updated `google.cloud.spanner_dbapi.connect()` to accept `username`
and `password` for DB-API connections.
- Updated sync and async transport factories to attach Bearer auth
interceptors to Omni gRPC channels.
* **Testing:** Added comprehensive unit test suites covering the OPAQUE
cryptographic engine, `LoginClient` state machine, credentials
lifecycle/interceptors, DB-API connect options, and system test
configurations for both sync and async client workflows against live
Spanner Omni instances.

---

### Running Integration Tests

To run integration tests against a live Spanner Omni instance with auth
login, set the following environment variables:

```bash
export SPANNER_OMNI=localhost:15000
export CA_CERTIFICATE=/path/to/ca.crt
export SPANNER_OMNI_USER=admin
export SPANNER_OMNI_PASSWORD=admin

python -m pytest -v -s --disable-warnings tests/system/ \
  -o asyncio_mode=auto \
  -o asyncio_default_test_loop_scope=session \
  -o asyncio_default_fixture_loop_scope=session
```
***

Author: sagnghos

packages/google-cloud-spanner/google/cloud/spanner_dbapi/connection.py

@@ -820,6 +820,8 @@ def connect(
     instance_type=None,
     data_boost_enabled=False,
     auto_partition_mode=False,
+    username=None,
+    password=None,
     **kwargs,
 ):
     """Creates a connection to a Google Cloud Spanner database.
@@ -909,6 +911,10 @@ def connect(
     :param client_key: (Optional) The path to the client key file used for mTLS connection.
         This is intended only for Spanner Omni endpoints.
         This is mandatory if Spanner Omni requires an mTLS connection.
+    :type username: str
+    :param username: (Optional) Username for Spanner Omni authentication.
+    :type password: str
+    :param password: (Optional) Password for Spanner Omni authentication.
     """
     if client is None:
         client_info = ClientInfo(
@@ -956,7 +962,29 @@ def connect(
                     )
 
                 project = "default"
-                credentials = AnonymousCredentials()
+                has_username = username is not None
+                has_password = password is not None
+                if has_username != has_password:
+                    raise ValueError(
+                        "Both username and password must be specified for Omni authentication"
+                    )
+                from google.cloud.spanner_v1.omni.credentials import (
+                    SpannerOmniCredentials,
+                )
+
+                if has_username and has_password:
+                    credentials = SpannerOmniCredentials(
+                        username=username,
+                        password=password,
+                        target=host_endpoint,
+                        use_plain_text=use_plain_text,
+                        ca_certificate=ca_certificate,
+                        client_certificate=client_certificate,
+                        client_key=client_key,
+                    )
+                else:
+                    credentials = AnonymousCredentials()
+
                 client_options = kwargs.get("client_options")
                 if client_options is None:
                     client_options = ClientOptions(api_endpoint=host_endpoint)
@@ -968,7 +996,6 @@ def connect(
 
                     client_options = copy.copy(client_options)
                     client_options.api_endpoint = host_endpoint
-
             client = spanner.Client(
                 project=project,
                 credentials=credentials,

packages/google-cloud-spanner/google/cloud/spanner_v1/_async/_helpers.py

@@ -75,6 +75,7 @@ def _create_spanner_omni_transport(
     client_certificate,
     client_key,
     interceptors=None,
+    credentials=None,
 ):
     """Creates a Spanner Omni transport in async mode.
 
@@ -87,6 +88,8 @@ def _create_spanner_omni_transport(
         client_certificate (str): Path to the client certificate file for mTLS.
         client_key (str): Path to the client key file for mTLS.
         interceptors (list): Optional list of interceptors to add to the channel.
+        credentials (google.auth.credentials.Credentials, optional): Credentials
+            to use for authentication.
 
     Returns:
         object: An instance of the transport class created by `transport_factory`.
@@ -98,8 +101,25 @@ def _create_spanner_omni_transport(
     from google.auth.credentials import AnonymousCredentials
 
     channel = None
+    all_interceptors = list(interceptors) if interceptors is not None else []
+    if credentials is not None:
+        if hasattr(credentials, "create_async_auth_interceptors"):
+            all_interceptors.extend(credentials.create_async_auth_interceptors())
+        elif hasattr(credentials, "create_async_auth_interceptor"):
+            res = credentials.create_async_auth_interceptor()
+            if isinstance(res, (list, tuple)):
+                all_interceptors.extend(res)
+            else:
+                all_interceptors.append(res)
+        elif hasattr(credentials, "create_auth_interceptor"):
+            res = credentials.create_auth_interceptor(is_async=True)
+            if isinstance(res, (list, tuple)):
+                all_interceptors.extend(res)
+            else:
+                all_interceptors.append(res)
+
     if use_plain_text:
-        channel = grpc.aio.insecure_channel(target=host, interceptors=interceptors)
+        channel = grpc.aio.insecure_channel(target=host, interceptors=all_interceptors)
     elif ca_certificate:
         with open(ca_certificate, "rb") as f:
             ca_cert = f.read()
@@ -119,12 +139,17 @@ def _create_spanner_omni_transport(
             )
         else:
             ssl_creds = grpc.ssl_channel_credentials(root_certificates=ca_cert)
-        channel = grpc.aio.secure_channel(host, ssl_creds, interceptors=interceptors)
+        channel = grpc.aio.secure_channel(
+            host, ssl_creds, interceptors=all_interceptors
+        )
     else:
         raise ValueError(
             "TLS/mTLS connection requires ca_certificate to be set for Spanner Omni"
         )
-    return transport_factory(channel=channel, credentials=AnonymousCredentials())
+    actual_credentials = (
+        credentials if credentials is not None else AnonymousCredentials()
+    )
+    return transport_factory(channel=channel, credentials=actual_credentials)
 
 
 def _create_experimental_host_transport(

packages/google-cloud-spanner/google/cloud/spanner_v1/_async/client.py

@@ -302,6 +302,8 @@ def __init__(
         client_certificate=None,
         client_key=None,
         instance_type=None,
+        username=None,
+        password=None,
     ):
         self._emulator_host = _get_spanner_emulator_host()
         self._use_plain_text = use_plain_text
@@ -353,10 +355,37 @@ def __init__(
             self._ca_certificate = ca_certificate
             self._client_certificate = client_certificate
             self._client_key = client_key
-            credentials = AnonymousCredentials()
+            self._host = host_endpoint
+            has_username = username is not None
+            has_password = password is not None
+            if has_username != has_password:
+                raise ValueError(
+                    "Both username and password must be specified for Omni authentication"
+                )
+            from google.cloud.spanner_v1.omni.credentials import (
+                SpannerOmniCredentials,
+            )
+
+            if has_username and has_password:
+                credentials = SpannerOmniCredentials(
+                    username=username,
+                    password=password,
+                    target=host_endpoint,
+                    use_plain_text=use_plain_text,
+                    ca_certificate=ca_certificate,
+                    client_certificate=client_certificate,
+                    client_key=client_key,
+                )
+            elif not isinstance(credentials, SpannerOmniCredentials):
+                credentials = AnonymousCredentials()
             disable_builtin_metrics = True
         elif isinstance(credentials, AnonymousCredentials):
             self._emulator_host = self._client_options.api_endpoint
+        else:
+            if username is not None or password is not None:
+                raise ValueError(
+                    "username and password can only be used when instance_type='omni'."
+                )
 
         # NOTE: This API has no use for the _http argument, but sending it
         #       will have no impact since the _http() @property only lazily
@@ -509,6 +538,7 @@ def instance_admin_api(self):
                         self._ca_certificate,
                         self._client_certificate,
                         self._client_key,
+                        credentials=self.credentials,
                     )
 
                 else:
@@ -519,6 +549,7 @@ def instance_admin_api(self):
                         self._ca_certificate,
                         self._client_certificate,
                         self._client_key,
+                        credentials=self.credentials,
                     )
 
                 self._instance_admin_api = InstanceAdminClient(
@@ -567,6 +598,7 @@ def database_admin_api(self):
                         self._ca_certificate,
                         self._client_certificate,
                         self._client_key,
+                        credentials=self.credentials,
                     )
 
                 else:
@@ -577,6 +609,7 @@ def database_admin_api(self):
                         self._ca_certificate,
                         self._client_certificate,
                         self._client_key,
+                        credentials=self.credentials,
                     )
 
                 self._database_admin_api = DatabaseAdminClient(

packages/google-cloud-spanner/google/cloud/spanner_v1/_async/database.py

@@ -513,6 +513,7 @@ def spanner_api(self):
                         client._ca_certificate,
                         client._client_certificate,
                         client._client_key,
+                        credentials=client.credentials,
                     )
                 else:
                     transport = _create_spanner_omni_transport_sync(
@@ -522,6 +523,7 @@ def spanner_api(self):
                         client._ca_certificate,
                         client._client_certificate,
                         client._client_key,
+                        credentials=client.credentials,
                     )
                 self._spanner_api = SpannerClient(
                     client_info=client_info,

packages/google-cloud-spanner/google/cloud/spanner_v1/_async/testing/database_test.py

@@ -138,6 +138,7 @@ def spanner_api(self):
                         client._client_certificate,
                         client._client_key,
                         self._interceptors,
+                        credentials=client.credentials,
                     )
                 else:
                     transport = _create_spanner_omni_transport_sync(
@@ -148,6 +149,7 @@ def spanner_api(self):
                         client._client_certificate,
                         client._client_key,
                         self._interceptors,
+                        credentials=client.credentials,
                     )
                 self._spanner_api = SpannerClient(
                     client_info=client_info,

packages/google-cloud-spanner/google/cloud/spanner_v1/_helpers.py

@@ -1044,6 +1044,7 @@ def _create_spanner_omni_transport(
     client_certificate,
     client_key,
     interceptors=None,
+    credentials=None,
 ):
     """Creates a Spanner Omni transport.
 
@@ -1056,6 +1057,8 @@ def _create_spanner_omni_transport(
         client_certificate (str): Path to the client certificate file for mTLS.
         client_key (str): Path to the client key file for mTLS.
         interceptors (list): Optional list of interceptors to add to the channel.
+        credentials (google.auth.credentials.Credentials, optional): Credentials
+            to use for authentication (e.g. `SpannerOmniCredentials`).
 
     Returns:
         object: An instance of the transport class created by `transport_factory`.
@@ -1067,6 +1070,10 @@ def _create_spanner_omni_transport(
     from google.auth.credentials import AnonymousCredentials
 
     channel = None
+    all_interceptors = list(interceptors) if interceptors is not None else []
+    if credentials is not None and hasattr(credentials, "create_auth_interceptor"):
+        all_interceptors.append(credentials.create_auth_interceptor())
+
     if use_plain_text:
         channel = grpc.insecure_channel(target=host)
     elif ca_certificate:
@@ -1093,9 +1100,12 @@ def _create_spanner_omni_transport(
         raise ValueError(
             "TLS/mTLS connection requires ca_certificate to be set for Spanner Omni"
         )
-    if interceptors is not None:
-        channel = grpc.intercept_channel(channel, *interceptors)
-    return transport_factory(channel=channel, credentials=AnonymousCredentials())
+    if all_interceptors:
+        channel = grpc.intercept_channel(channel, *all_interceptors)
+    actual_credentials = (
+        credentials if credentials is not None else AnonymousCredentials()
+    )
+    return transport_factory(channel=channel, credentials=actual_credentials)
 
 
 def _create_experimental_host_transport(

packages/google-cloud-spanner/google/cloud/spanner_v1/client.py

@@ -267,6 +267,8 @@ def __init__(
         client_certificate=None,
         client_key=None,
         instance_type=None,
+        username=None,
+        password=None,
     ):
         self._emulator_host = _get_spanner_emulator_host()
         self._use_plain_text = use_plain_text
@@ -316,10 +318,37 @@ def __init__(
             self._ca_certificate = ca_certificate
             self._client_certificate = client_certificate
             self._client_key = client_key
-            credentials = AnonymousCredentials()
+            self._host = host_endpoint
+            has_username = username is not None
+            has_password = password is not None
+            if has_username != has_password:
+                raise ValueError(
+                    "Both username and password must be specified for Omni authentication"
+                )
+            from google.cloud.spanner_v1.omni.credentials import (
+                SpannerOmniCredentials,
+            )
+
+            if has_username and has_password:
+                credentials = SpannerOmniCredentials(
+                    username=username,
+                    password=password,
+                    target=host_endpoint,
+                    use_plain_text=use_plain_text,
+                    ca_certificate=ca_certificate,
+                    client_certificate=client_certificate,
+                    client_key=client_key,
+                )
+            elif not isinstance(credentials, SpannerOmniCredentials):
+                credentials = AnonymousCredentials()
             disable_builtin_metrics = True
         elif isinstance(credentials, AnonymousCredentials):
             self._emulator_host = self._client_options.api_endpoint
+        else:
+            if username is not None or password is not None:
+                raise ValueError(
+                    "username and password can only be used when instance_type='omni'."
+                )
         super(Client, self).__init__(
             project=project,
             credentials=credentials,
@@ -443,6 +472,7 @@ def instance_admin_api(self):
                     self._ca_certificate,
                     self._client_certificate,
                     self._client_key,
+                    credentials=self.credentials,
                 )
                 self._instance_admin_api = InstanceAdminClient(
                     client_info=self._client_info,
@@ -481,6 +511,7 @@ def database_admin_api(self):
                     self._ca_certificate,
                     self._client_certificate,
                     self._client_key,
+                    credentials=self.credentials,
                 )
                 self._database_admin_api = DatabaseAdminClient(
                     client_info=self._client_info,

packages/google-cloud-spanner/google/cloud/spanner_v1/database.py

@@ -452,6 +452,7 @@ def spanner_api(self):
                     client._ca_certificate,
                     client._client_certificate,
                     client._client_key,
+                    credentials=client.credentials,
                 )
                 self._spanner_api = SpannerClient(
                     client_info=client_info,

packages/google-cloud-spanner/google/cloud/spanner_v1/omni/__init__.py

@@ -0,0 +1,27 @@
+# -*- coding: utf-8 -*-
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""Spanner Omni authentication and connection utilities."""
+
+from google.cloud.spanner_v1.omni.credentials import SpannerOmniCredentials
+from google.cloud.spanner_v1.omni.login_client import LoginClient
+from google.cloud.spanner_v1.omni.opaque import UserAuthenticator
+
+__all__ = (
+    "LoginClient",
+    "SpannerOmniCredentials",
+    "UserAuthenticator",
+)