PYCBC-1714: Support mTLS Certs Refresh (without restart)

Changes
=======

- Add update_credentials to sync (couchbase) and async (acouchbase) APIs
- Wire through Python/C++ bindings to refresh TLS context without restart
- Update tests for credential refresh path

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

Author: anirudhlakhotia

acouchbase/cluster.py

@@ -22,7 +22,8 @@
 from typing import (TYPE_CHECKING,
                     Any,
                     Awaitable,
-                    Dict)
+                    Dict,
+                    Union)
 
 from acouchbase import get_event_loop
 from acouchbase.analytics import AnalyticsQuery, AsyncAnalyticsRequest
@@ -37,6 +38,7 @@
 from acouchbase.n1ql import AsyncN1QLRequest, N1QLQuery
 from acouchbase.search import AsyncFullTextSearchRequest, SearchQueryBuilder
 from acouchbase.transactions import Transactions
+from couchbase.auth import CertificateAuthenticator, PasswordAuthenticator
 from couchbase.diagnostics import ClusterState, ServiceType
 from couchbase.exceptions import UnAmbiguousTimeoutException
 from couchbase.logic.cluster import ClusterLogic
@@ -179,6 +181,19 @@ async def close(self) -> None:
         await self._close_ftr
         super()._destroy_connection()
 
+    def update_credentials(self, authenticator: Union[CertificateAuthenticator, PasswordAuthenticator]) -> None:
+        """Update the credentials used by this Cluster.
+
+        Args:
+            authenticator (Union[CertificateAuthenticator, PasswordAuthenticator]): New authenticator.
+        """
+        if not self.connected:
+            raise RuntimeError("Cluster is not connected, cannot update credentials.")
+
+        # This is a fast, synchronous operation in the core; call directly.
+        super()._update_credentials(auth=authenticator.as_dict())
+        self._auth = authenticator.as_dict()
+
     def bucket(self, bucket_name) -> AsyncBucket:
         """Creates a Bucket instance to a specific bucket.
 

acouchbase/tests/credentials_t.py

@@ -0,0 +1,78 @@
+#  Copyright 2016-2025. Couchbase, Inc.
+#  All Rights Reserved.
+#
+#  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.
+
+import uuid
+
+import pytest
+import pytest_asyncio
+
+from acouchbase.cluster import Cluster as AsyncCluster
+from couchbase.auth import CertificateAuthenticator, PasswordAuthenticator
+from couchbase.exceptions import InvalidArgumentException
+from couchbase.options import ClusterOptions
+
+
+class AsyncCredentialsTests:
+
+    @pytest_asyncio.fixture(scope="class")
+    async def cb_env(self, couchbase_config):
+        class Env:
+            def __init__(self, cfg):
+                self.cfg = cfg
+                self.cluster = None
+        env = Env(couchbase_config)
+        conn_string = couchbase_config.get_connection_string()
+        username, pw = couchbase_config.get_username_and_pw()
+        env.cluster = await AsyncCluster.connect(conn_string, ClusterOptions(PasswordAuthenticator(username, pw)))
+        yield env
+        await env.cluster.close()
+
+    @pytest.mark.asyncio
+    async def test_update_credentials_async(self, cb_env):
+        cluster = cb_env.cluster
+
+        new_user = f"pycbc_{uuid.uuid4().hex[:8]}"
+        new_pass = f"pw_{uuid.uuid4().hex[:8]}"
+        cluster.update_credentials(PasswordAuthenticator(new_user, new_pass))
+
+        info_after = cluster._get_client_connection_info()
+        assert info_after['credentials']['username'] == new_user
+        assert info_after['credentials']['password'] == new_pass
+
+    @pytest.mark.asyncio
+    async def test_update_to_certificate_auth_without_tls_fails_async(self, cb_env):
+        cluster = cb_env.cluster
+        # Attempt to switch to certificate auth without TLS should raise
+        with pytest.raises(InvalidArgumentException):
+            cluster.update_credentials(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):
+        cluster = cb_env.cluster
+
+        # capture original
+        info_before = cluster._get_client_connection_info()
+        assert 'credentials' in info_before
+        orig_creds = info_before['credentials']
+
+        # 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'))
+
+        # ensure credentials are unchanged after the failed update
+        info_after = cluster._get_client_connection_info()
+        assert info_after['credentials'] == orig_creds

couchbase/cluster.py

@@ -19,9 +19,11 @@
 from datetime import timedelta
 from typing import (TYPE_CHECKING,
                     Any,
-                    Dict)
+                    Dict,
+                    Union)
 
 from couchbase.analytics import AnalyticsQuery, AnalyticsRequest
+from couchbase.auth import CertificateAuthenticator, PasswordAuthenticator
 from couchbase.bucket import Bucket
 from couchbase.diagnostics import ClusterState, ServiceType
 from couchbase.exceptions import ErrorMapper, UnAmbiguousTimeoutException
@@ -229,6 +231,17 @@ def diagnostics(self,
 
         return super().diagnostics(*opts, **kwargs)
 
+    def update_credentials(self, authenticator: Union[CertificateAuthenticator, PasswordAuthenticator]) -> None:
+        """Update the credentials used by this Cluster.
+
+        Args:
+            authenticator (Union[CertificateAuthenticator, PasswordAuthenticator]): New authenticator.
+        """
+        if not self.connected:
+            raise RuntimeError("Cluster not yet connected.")
+        super()._update_credentials(auth=authenticator.as_dict())
+        self._auth = authenticator.as_dict()
+
     def wait_until_ready(self,
                          timeout,  # type: timedelta
                          *opts,  # type: WaitUntilReadyOptions

couchbase/logic/cluster.py

@@ -29,7 +29,8 @@
 from couchbase import USER_AGENT_EXTRA
 from couchbase.auth import CertificateAuthenticator, PasswordAuthenticator
 from couchbase.diagnostics import ServiceType
-from couchbase.exceptions import InvalidArgumentException
+from couchbase.exceptions import ErrorMapper, InvalidArgumentException
+from couchbase.exceptions import exception as BaseCouchbaseException
 from couchbase.options import (ClusterMetricsOptions,
                                ClusterOptions,
                                ClusterOrphanReportingOptions,
@@ -46,7 +47,8 @@
                                   get_connection_info,
                                   management_operation,
                                   mgmt_operations,
-                                  operations)
+                                  operations,
+                                  update_credentials)
 from couchbase.result import (ClusterInfoResult,
                               DiagnosticsResult,
                               PingResult)
@@ -471,6 +473,22 @@ def _close_cluster(self, **kwargs):
             self._connection, **close_kwargs
         )
 
+    def _update_credentials(self, **kwargs):
+        """**INTERNAL** not intended for use in public API.
+
+        Expects kwargs to contain:
+          - auth: dict of authenticator values (see couchbase.auth.Authenticator.as_dict())
+        """
+
+        auth = kwargs.pop('auth', None)
+        if auth is None:
+            raise InvalidArgumentException('Authenticator (auth) must be provided.')
+
+        ret = update_credentials(self._connection, auth=auth)
+        if isinstance(ret, BaseCouchbaseException):
+            raise ErrorMapper.build_exception(ret)
+        return ret
+
     def _set_connection(self, conn):
         self._connection = conn
 

couchbase/tests/credentials_t.py

@@ -0,0 +1,88 @@
+#  Copyright 2016-2025. Couchbase, Inc.
+#  All Rights Reserved.
+#
+#  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.
+
+import uuid
+
+import pytest
+
+from couchbase.auth import CertificateAuthenticator, PasswordAuthenticator
+from couchbase.cluster import Cluster
+from couchbase.exceptions import InvalidArgumentException
+from couchbase.options import ClusterOptions
+
+
+class ClassicCredentialsTests:
+
+    def test_update_credentials_reflected_in_connection_info(self, couchbase_config):
+        conn_string = couchbase_config.get_connection_string()
+        username, pw = couchbase_config.get_username_and_pw()
+
+        cluster = Cluster.connect(conn_string, ClusterOptions(PasswordAuthenticator(username, pw)))
+
+        # capture original
+        info_before = cluster._get_client_connection_info()
+        assert 'credentials' in info_before
+        orig_creds = info_before['credentials']
+
+        # update to new (likely invalid) 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))
+
+        info_after = cluster._get_client_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.close()
+
+    def test_update_to_certificate_auth_without_tls_fails(self, couchbase_config):
+        conn_string = couchbase_config.get_connection_string()
+        username, pw = couchbase_config.get_username_and_pw()
+
+        # Non-TLS connection (expected couchbase://)
+        cluster = Cluster.connect(conn_string, ClusterOptions(PasswordAuthenticator(username, pw)))
+
+        # 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.close()
+
+    def test_update_credentials_failure_does_not_change_state(self, couchbase_config):
+        conn_string = couchbase_config.get_connection_string()
+        username, pw = couchbase_config.get_username_and_pw()
+
+        cluster = Cluster.connect(conn_string, ClusterOptions(PasswordAuthenticator(username, pw)))
+
+        # capture original
+        info_before = cluster._get_client_connection_info()
+        assert 'credentials' in info_before
+        orig_creds = info_before['credentials']
+
+        # 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'))
+
+        # ensure credentials are unchanged after the failed update
+        info_after = cluster._get_client_connection_info()
+        assert info_after['credentials'] == orig_creds
+
+        cluster.close()

deps/couchbase-cxx-client

@@ -300,6 +300,17 @@ create_connection(PyObject* self, PyObject* args, PyObject* kwargs)
   return res;
 }
 
+static PyObject*
+update_credentials(PyObject* self, PyObject* args, PyObject* kwargs)
+{
+  PyObject* res = handle_update_credentials(self, args, kwargs);
+  if (res == nullptr && PyErr_Occurred() == nullptr) {
+    pycbc_set_python_exception(
+      PycbcError::UnsuccessfulOperation, __FILE__, __LINE__, "Unable to update credentials.");
+  }
+  return res;
+}
+
 static PyObject*
 get_connection_information(PyObject* self, PyObject* args, PyObject* kwargs)
 {
@@ -336,6 +347,10 @@ static struct PyMethodDef methods[] = {
     (PyCFunction)create_connection,
     METH_VARARGS | METH_KEYWORDS,
     "Create connection object" },
+  { "update_credentials",
+    (PyCFunction)update_credentials,
+    METH_VARARGS | METH_KEYWORDS,
+    "Update connection credentials" },
   { "get_connection_info",
     (PyCFunction)get_connection_information,
     METH_VARARGS | METH_KEYWORDS,

src/client.cxx

@@ -1565,3 +1565,43 @@ handle_open_or_close_bucket([[maybe_unused]] PyObject* self, PyObject* args, PyO
   }
   Py_RETURN_NONE;
 }
+
+PyObject*
+handle_update_credentials([[maybe_unused]] PyObject* self, PyObject* args, PyObject* kwargs)
+{
+  PyObject* pyObj_conn = nullptr;
+  PyObject* pyObj_auth = nullptr;
+
+  static const char* kw_list[] = { "", "auth", nullptr };
+
+  const char* kw_format = "O!O!";
+  int ret = PyArg_ParseTupleAndKeywords(args,
+                                        kwargs,
+                                        kw_format,
+                                        const_cast<char**>(kw_list),
+                                        &PyCapsule_Type,
+                                        &pyObj_conn,
+                                        &PyDict_Type,
+                                        &pyObj_auth);
+
+  if (!ret) {
+    std::string msg = "Cannot update credentials. Unable to parse args/kwargs.";
+    pycbc_set_python_exception(PycbcError::InvalidArgument, __FILE__, __LINE__, msg.c_str());
+    return nullptr;
+  }
+
+  connection* conn = reinterpret_cast<connection*>(PyCapsule_GetPointer(pyObj_conn, "conn_"));
+  if (nullptr == conn) {
+    pycbc_set_python_exception(PycbcError::InvalidArgument, __FILE__, __LINE__, NULL_CONN_OBJECT);
+    return nullptr;
+  }
+
+  couchbase::core::cluster_credentials auth = get_cluster_credentials(pyObj_auth);
+
+  auto err = conn->cluster_.update_credentials(std::move(auth));
+  if (err.ec) {
+    return pycbc_build_exception(err.ec, __FILE__, __LINE__, err.message);
+  }
+
+  Py_RETURN_NONE;
+}

src/connection.cxx

@@ -30,3 +30,6 @@ handle_close_connection(PyObject* self, PyObject* args, PyObject* kwargs);
 
 PyObject*
 handle_open_or_close_bucket(PyObject* self, PyObject* args, PyObject* kwargs);
+
+PyObject*
+handle_update_credentials(PyObject* self, PyObject* args, PyObject* kwargs);