PYCBC-1715: Support JWT based authentication in Operational SDKs

Change-Id: I6b19ed9def93edaa08844fe258c9ac9c35b3c469
Reviewed-on: https://review.couchbase.org/c/couchbase-python-client/+/239443
Reviewed-by: Jared Casey <jared.casey@couchbase.com>
Tested-by: Build Bot <build@couchbase.com>

Author: anirudhlakhotia

acouchbase/cluster.py

@@ -32,7 +32,9 @@
 from acouchbase.management.search import SearchIndexManager
 from acouchbase.management.users import UserManager
 from acouchbase.transactions import Transactions
-from couchbase.auth import CertificateAuthenticator, PasswordAuthenticator
+from couchbase.auth import (CertificateAuthenticator,
+                            JwtAuthenticator,
+                            PasswordAuthenticator)
 from couchbase.result import (AnalyticsResult,
                               ClusterInfoResult,
                               DiagnosticsResult,
@@ -129,13 +131,29 @@ async def close(self) -> None:
         """
         await self._impl.close_connection()
 
-    def update_credentials(self, authenticator: Union[CertificateAuthenticator, PasswordAuthenticator]) -> None:
-        """Update the credentials used by this Cluster.
+    def set_authenticator(
+            self, authenticator: Union[CertificateAuthenticator, JwtAuthenticator, PasswordAuthenticator]
+    ) -> None:
+        """Replace the authenticator used by this Cluster.
+
+        Allows updating credentials without restarting the application.
+        The effect on existing connections depends on the authenticator type:
+
+        - JwtAuthenticator: Live re-auth on existing KV connections via OAUTHBEARER SASL.
+          HTTP requests use the new Bearer token immediately.
+        - PasswordAuthenticator: No effect on existing KV connections (they keep old
+          credentials). New HTTP requests use the new Basic auth header.
+        - CertificateAuthenticator: No effect on existing connections (TLS handshake
+          already completed). Only new connections use the new certificate.
 
         Args:
-            authenticator (Union[CertificateAuthenticator, PasswordAuthenticator]): New authenticator.
+            authenticator: New authenticator to use. Must be the same type as the
+                current authenticator.
+
+        Raises:
+            RuntimeError: If cluster is not connected.
         """
-        req = self._impl.request_builder.build_udpate_credential_request(authenticator)
+        req = self._impl.request_builder.build_update_credential_request(authenticator)
         self._impl.update_credentials(req)
 
     def bucket(self, bucket_name: str) -> AsyncBucket:

acouchbase/tests/credentials_t.py

@@ -19,11 +19,34 @@
 import pytest_asyncio
 
 from acouchbase.cluster import Cluster as AsyncCluster
-from couchbase.auth import CertificateAuthenticator, PasswordAuthenticator
+from couchbase.auth import (CertificateAuthenticator,
+                            JwtAuthenticator,
+                            PasswordAuthenticator)
 from couchbase.exceptions import InvalidArgumentException
 from couchbase.options import ClusterOptions
 
 
+class JwtAuthenticatorUnitTests:
+    """Unit tests for JwtAuthenticator that don't require a cluster connection."""
+
+    def test_jwt_authenticator_creation(self):
+        token = "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.test.signature"
+        auth = JwtAuthenticator(token)
+        assert auth.as_dict() == {'jwt_token': token}
+
+    def test_jwt_authenticator_valid_keys(self):
+        auth = JwtAuthenticator("test.jwt.token")
+        assert auth.valid_keys() == ['jwt_token']
+
+    def test_jwt_authenticator_rejects_non_string(self):
+        with pytest.raises(InvalidArgumentException):
+            JwtAuthenticator(12345)
+
+    def test_jwt_authenticator_rejects_none(self):
+        with pytest.raises(InvalidArgumentException):
+            JwtAuthenticator(None)
+
+
 class AsyncCredentialsTests:
 
     @pytest_asyncio.fixture(scope="class")
@@ -40,27 +63,37 @@ def __init__(self, cfg):
         await env.cluster.close()
 
     @pytest.mark.asyncio
-    async def test_update_credentials_async(self, cb_env):
+    async def test_set_authenticator_reflected_in_connection_info_async(self, cb_env):
         cluster = cb_env.cluster
 
+        # capture original
+        info_before = cluster._impl.get_connection_info()
+        assert 'credentials' in info_before
+        orig_creds = info_before['credentials']
+
+        # update to new creds; we only assert that core origin is updated
         new_user = f"pycbc_{uuid.uuid4().hex[:8]}"
         new_pass = f"pw_{uuid.uuid4().hex[:8]}"
-        cluster.update_credentials(PasswordAuthenticator(new_user, new_pass))
+        cluster.set_authenticator(PasswordAuthenticator(new_user, new_pass))
 
         info_after = cluster._impl.get_connection_info()
         assert info_after['credentials']['username'] == new_user
         assert info_after['credentials']['password'] == new_pass
 
+        # restore original to avoid impacting other tests
+        cluster.set_authenticator(PasswordAuthenticator(orig_creds['username'],
+                                                        orig_creds['password']))
+
     @pytest.mark.asyncio
-    async def test_update_to_certificate_auth_without_tls_fails_async(self, cb_env):
+    async def test_set_authenticator_password_to_certificate_fails_async(self, cb_env):
         cluster = cb_env.cluster
-        # Attempt to switch to certificate auth without TLS should raise
+        # Core should reject this at validation step, surfacing as InvalidArgumentException
         with pytest.raises(InvalidArgumentException):
-            cluster.update_credentials(CertificateAuthenticator(cert_path='path/to/cert',
-                                                                key_path='path/to/key'))
+            cluster.set_authenticator(CertificateAuthenticator(cert_path='path/to/cert',
+                                                               key_path='path/to/key'))
 
     @pytest.mark.asyncio
-    async def test_update_credentials_failure_does_not_change_state_async(self, cb_env):
+    async def test_set_authenticator_failure_does_not_change_state_async(self, cb_env):
         cluster = cb_env.cluster
 
         # capture original
@@ -70,9 +103,18 @@ async def test_update_credentials_failure_does_not_change_state_async(self, cb_e
 
         # attempt to switch to certificate auth on non-TLS connection; expect failure
         with pytest.raises(InvalidArgumentException):
-            cluster.update_credentials(CertificateAuthenticator(cert_path='path/to/cert',
-                                                                key_path='path/to/key'))
+            cluster.set_authenticator(CertificateAuthenticator(cert_path='path/to/cert',
+                                                               key_path='path/to/key'))
 
         # ensure credentials are unchanged after the failed update
         info_after = cluster._impl.get_connection_info()
         assert info_after['credentials'] == orig_creds
+
+    @pytest.mark.asyncio
+    async def test_set_authenticator_password_to_jwt_fails_async(self, cb_env):
+        """RFC: Cannot switch from PasswordAuthenticator to JwtAuthenticator."""
+        cluster = cb_env.cluster
+
+        # RFC: Cannot switch authenticator types
+        with pytest.raises(InvalidArgumentException):
+            cluster.set_authenticator(JwtAuthenticator("some.jwt.token"))

couchbase/auth.py

@@ -100,6 +100,40 @@ def ldap_compatible(username,  # type: str
         return auth
 
 
+class JwtAuthenticator(Authenticator):
+    """
+    JWT (JSON Web Token) authentication mechanism.
+
+    Uses OAUTHBEARER SASL mechanism for KV connections and Bearer token
+    for HTTP services.
+
+    Args:
+        token (str): The JWT token to use for authentication.
+
+    Raises:
+        :class:`~couchbase.exceptions.InvalidArgumentException`: If token is not a string.
+    """
+
+    def __init__(self, token  # type: str
+                 ):
+        """JwtAuthenticator instance."""
+        if not isinstance(token, str):
+            msg = 'The token must be a str.'
+            raise InvalidArgumentException(msg)
+
+        self._token = token
+
+        super().__init__(**self.as_dict())
+
+    def valid_keys(self):
+        return ['jwt_token']
+
+    def as_dict(self):
+        return {
+            'jwt_token': self._token
+        }
+
+
 class CertificateAuthenticator(Authenticator):
     """
     Certificate authentication mechanism.

couchbase/cluster.py

@@ -21,7 +21,9 @@
                     Dict,
                     Union)
 
-from couchbase.auth import CertificateAuthenticator, PasswordAuthenticator
+from couchbase.auth import (CertificateAuthenticator,
+                            JwtAuthenticator,
+                            PasswordAuthenticator)
 from couchbase.bucket import Bucket
 from couchbase.logic.cluster_impl import ClusterImpl
 from couchbase.logic.supportability import Supportability
@@ -191,13 +193,29 @@ def diagnostics(self,
         req = self._impl.request_builder.build_diagnostics_request(*opts, **kwargs)
         return self._impl.diagnostics(req)
 
-    def update_credentials(self, authenticator: Union[CertificateAuthenticator, PasswordAuthenticator]) -> None:
-        """Update the credentials used by this Cluster.
+    def set_authenticator(
+            self, authenticator: Union[CertificateAuthenticator, JwtAuthenticator, PasswordAuthenticator]
+    ) -> None:
+        """Replace the authenticator used by this Cluster.
+
+        Allows updating credentials without restarting the application.
+        The effect on existing connections depends on the authenticator type:
+
+        - JwtAuthenticator: Live re-auth on existing KV connections via OAUTHBEARER SASL.
+          HTTP requests use the new Bearer token immediately.
+        - PasswordAuthenticator: No effect on existing KV connections (they keep old
+          credentials). New HTTP requests use the new Basic auth header.
+        - CertificateAuthenticator: No effect on existing connections (TLS handshake
+          already completed). Only new connections use the new certificate.
 
         Args:
-            authenticator (Union[CertificateAuthenticator, PasswordAuthenticator]): New authenticator.
+            authenticator: New authenticator to use. Must be the same type as the
+                current authenticator.
+
+        Raises:
+            RuntimeError: If cluster is not connected.
         """
-        req = self._impl.request_builder.build_udpate_credential_request(authenticator)
+        req = self._impl.request_builder.build_update_credential_request(authenticator)
         self._impl.update_credentials(req)
 
     def wait_until_ready(self,

couchbase/logic/cluster_req_builder.py

@@ -37,7 +37,9 @@
                               SearchRequest)
 
 if TYPE_CHECKING:
-    from couchbase.auth import CertificateAuthenticator, PasswordAuthenticator
+    from couchbase.auth import (CertificateAuthenticator,
+                                JwtAuthenticator,
+                                PasswordAuthenticator)
 
 
 class ClusterRequestBuilder:
@@ -113,8 +115,9 @@ def build_search_request(self,
             req.num_workers = num_workers
         return req
 
-    def build_udpate_credential_request(
-            self, authenticator: Union[CertificateAuthenticator, PasswordAuthenticator]) -> UpdateCredentialsRequest:
+    def build_update_credential_request(
+            self, authenticator: Union[CertificateAuthenticator, JwtAuthenticator, PasswordAuthenticator]
+    ) -> UpdateCredentialsRequest:
         return UpdateCredentialsRequest(authenticator.as_dict())
 
     def build_wait_until_ready_request(self,

couchbase/tests/credentials_t.py

@@ -17,15 +17,38 @@
 
 import pytest
 
-from couchbase.auth import CertificateAuthenticator, PasswordAuthenticator
+from couchbase.auth import (CertificateAuthenticator,
+                            JwtAuthenticator,
+                            PasswordAuthenticator)
 from couchbase.cluster import Cluster
 from couchbase.exceptions import InvalidArgumentException
 from couchbase.options import ClusterOptions
 
 
+class JwtAuthenticatorUnitTests:
+    """Unit tests for JwtAuthenticator that don't require a cluster connection."""
+
+    def test_jwt_authenticator_creation(self):
+        token = "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.test.signature"
+        auth = JwtAuthenticator(token)
+        assert auth.as_dict() == {'jwt_token': token}
+
+    def test_jwt_authenticator_valid_keys(self):
+        auth = JwtAuthenticator("test.jwt.token")
+        assert auth.valid_keys() == ['jwt_token']
+
+    def test_jwt_authenticator_rejects_non_string(self):
+        with pytest.raises(InvalidArgumentException):
+            JwtAuthenticator(12345)
+
+    def test_jwt_authenticator_rejects_none(self):
+        with pytest.raises(InvalidArgumentException):
+            JwtAuthenticator(None)
+
+
 class ClassicCredentialsTests:
 
-    def test_update_credentials_reflected_in_connection_info(self, couchbase_config):
+    def test_set_authenticator_reflected_in_connection_info(self, couchbase_config):
         conn_string = couchbase_config.get_connection_string()
         username, pw = couchbase_config.get_username_and_pw()
 
@@ -36,22 +59,22 @@ def test_update_credentials_reflected_in_connection_info(self, couchbase_config)
         assert 'credentials' in info_before
         orig_creds = info_before['credentials']
 
-        # update to new (likely invalid) creds; we only assert that core origin is updated
+        # update to new creds; we only assert that core origin is updated
         new_user = f"pycbc_{uuid.uuid4().hex[:8]}"
         new_pass = f"pw_{uuid.uuid4().hex[:8]}"
-        cluster.update_credentials(PasswordAuthenticator(new_user, new_pass))
+        cluster.set_authenticator(PasswordAuthenticator(new_user, new_pass))
 
         info_after = cluster._impl.get_connection_info()
         assert info_after['credentials']['username'] == new_user
         assert info_after['credentials']['password'] == new_pass
 
         # restore original to avoid impacting other tests
-        cluster.update_credentials(PasswordAuthenticator(orig_creds.get('username', username),
-                                                         orig_creds.get('password', pw)))
+        cluster.set_authenticator(PasswordAuthenticator(orig_creds.get('username', username),
+                                                        orig_creds.get('password', pw)))
 
         cluster.close()
 
-    def test_update_to_certificate_auth_without_tls_fails(self, couchbase_config):
+    def test_set_authenticator_password_to_certificate_fails(self, couchbase_config):
         conn_string = couchbase_config.get_connection_string()
         username, pw = couchbase_config.get_username_and_pw()
 
@@ -60,12 +83,12 @@ def test_update_to_certificate_auth_without_tls_fails(self, couchbase_config):
 
         # Core should reject this at validation step, surfacing as InvalidArgumentException
         with pytest.raises(InvalidArgumentException):
-            cluster.update_credentials(CertificateAuthenticator(cert_path='path/to/cert',
-                                                                key_path='path/to/key'))
+            cluster.set_authenticator(CertificateAuthenticator(cert_path='path/to/cert',
+                                                               key_path='path/to/key'))
 
         cluster.close()
 
-    def test_update_credentials_failure_does_not_change_state(self, couchbase_config):
+    def test_set_authenticator_failure_does_not_change_state(self, couchbase_config):
         conn_string = couchbase_config.get_connection_string()
         username, pw = couchbase_config.get_username_and_pw()
 
@@ -78,11 +101,24 @@ def test_update_credentials_failure_does_not_change_state(self, couchbase_config
 
         # attempt to switch to certificate auth on non-TLS connection; expect failure
         with pytest.raises(InvalidArgumentException):
-            cluster.update_credentials(CertificateAuthenticator(cert_path='path/to/cert',
-                                                                key_path='path/to/key'))
+            cluster.set_authenticator(CertificateAuthenticator(cert_path='path/to/cert',
+                                                               key_path='path/to/key'))
 
         # ensure credentials are unchanged after the failed update
         info_after = cluster._impl.get_connection_info()
         assert info_after['credentials'] == orig_creds
 
         cluster.close()
+
+    def test_set_authenticator_password_to_jwt_fails(self, couchbase_config):
+        """RFC: Cannot switch from PasswordAuthenticator to JwtAuthenticator."""
+        conn_string = couchbase_config.get_connection_string()
+        username, pw = couchbase_config.get_username_and_pw()
+
+        cluster = Cluster.connect(conn_string, ClusterOptions(PasswordAuthenticator(username, pw)))
+
+        # RFC: Cannot switch authenticator types
+        with pytest.raises(InvalidArgumentException):
+            cluster.set_authenticator(JwtAuthenticator("some.jwt.token"))
+
+        cluster.close()

src/cpp_core_types.hxx

@@ -159,6 +159,7 @@ struct py_to_cbpp_t<couchbase::core::cluster_credentials> {
     extract_field(pyObj, "cert_path", creds.certificate_path);
     extract_field(pyObj, "key_path", creds.key_path);
     extract_field(pyObj, "allowed_sasl_mechanisms", creds.allowed_sasl_mechanisms);
+    extract_field(pyObj, "jwt_token", creds.jwt_token);
 
     return creds;
   }
@@ -174,6 +175,7 @@ struct py_to_cbpp_t<couchbase::core::cluster_credentials> {
     add_string_field_if_not_empty(dict, "password", creds.password);
     add_string_field_if_not_empty(dict, "cert_path", creds.certificate_path);
     add_string_field_if_not_empty(dict, "key_path", creds.key_path);
+    add_string_field_if_not_empty(dict, "jwt_token", creds.jwt_token);
 
     if (creds.allowed_sasl_mechanisms.has_value() &&
         !creds.allowed_sasl_mechanisms.value().empty()) {