From 374de81854b10961c0c115339af223263be9b9e6 Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Fri, 4 Sep 2026 15:40:02 -0400 Subject: [PATCH 01/22] CI: Bump pinned BlazingMQ tag to v0.95.20 CI failed to compile `pybmq_session.cpp`: `AuthnCredentialCbFunctor` is called with no arguments, but `bmqt::SessionOptions::AuthnCredentialCb` at the pinned `v0.95.14` is still bsl::function(bsl::ostream& error)> The `ostream& error` parameter was removed upstream in bloomberg/blazingmq@c1614448a1c9356200e22cfa59cf623ff187dc0c ("Refactor: User authentication credentials callback (#1571)"), first released in v0.95.15. Our code was written against that signature, not the one the pinned tag actually provides. `bmqt_authncredential.h` (the `AuthnCredential` value type itself) is unchanged between v0.95.14 and v0.95.20, and no other commit in that range touches `SessionOptions`'s public surface except an internal allocator fix (c0f272850), so nothing else in this branch needs to change for the bump. Bump to v0.95.20, the latest tag, rather than the minimal v0.95.15, since CMakeLists.txt is untouched across the whole range and BDE_TAG/NTF_CORE_TAG need no corresponding change. --- bin/clone-dependencies.sh | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/bin/clone-dependencies.sh b/bin/clone-dependencies.sh index 10260a8..1949ecb 100755 --- a/bin/clone-dependencies.sh +++ b/bin/clone-dependencies.sh @@ -15,7 +15,7 @@ set -u BDE_TOOLS_TAG=4.38.0.0 BDE_TAG=4.38.0.0 NTF_CORE_TAG=2.6.12 -BLAZINGMQ_TAG=v0.95.14 +BLAZINGMQ_TAG=v0.95.20 if [ ! -d "${DIR_THIRDPARTY}/bde-tools" ]; then From 09317c374e5d2d9e6a38a7d2c19b8c420cbc5259 Mon Sep 17 00:00:00 2001 From: Emelia Lei Date: Mon, 8 Dec 2025 17:25:10 -0500 Subject: [PATCH 02/22] provide a way to create credential Signed-off-by: Emelia Lei --- src/blazingmq/_ext.pyi | 4 ++ src/blazingmq/_ext.pyx | 35 ++++++++++++++++ src/blazingmq/_session.py | 22 ++++++++++ src/cpp/pybmq_session.cpp | 79 ++++++++++++++++++++++++++++++++++- src/cpp/pybmq_session.h | 6 +++ src/declarations/bmq/bmqt.pxd | 10 +++++ src/declarations/pybmq.pxd | 1 + 7 files changed, 155 insertions(+), 2 deletions(-) diff --git a/src/blazingmq/_ext.pyi b/src/blazingmq/_ext.pyi index b455d49..de49aa7 100644 --- a/src/blazingmq/_ext.pyi +++ b/src/blazingmq/_ext.pyi @@ -37,6 +37,9 @@ class FakeHostHealthMonitor: def set_healthy(self) -> None: ... def set_unhealthy(self) -> None: ... +class FakeAuthnCredentialCb: + def __init__(self, callback: Callable[[], Optional[tuple[str, bytes]]]) -> None: ... + class Session: def __init__( self, @@ -53,6 +56,7 @@ class Session: timeouts: Timeouts = Timeouts(), monitor_host_health: bool = False, fake_host_health_monitor: Optional[FakeHostHealthMonitor] = None, + fake_authn_credential_cb: Optional[FakeAuthnCredentialCb] = None, user_agent_prefix: bytes = b"", ) -> None: ... def stop(self) -> None: ... diff --git a/src/blazingmq/_ext.pyx b/src/blazingmq/_ext.pyx index 4d07e74..f1b0d8e 100644 --- a/src/blazingmq/_ext.pyx +++ b/src/blazingmq/_ext.pyx @@ -21,12 +21,15 @@ import weakref from bsl cimport optional from bsl cimport pair from bsl cimport shared_ptr +from bsl cimport vector +from bsl cimport string from bsl.bsls cimport TimeInterval from cpython.ceval cimport PyEval_InitThreads from libcpp cimport bool as cppbool from bmq.bmqa cimport ManualHostHealthMonitor from bmq.bmqt cimport AckResult +from bmq.bmqt cimport AuthnCredential from bmq.bmqt cimport CompressionAlgorithmType from bmq.bmqt cimport HostHealthState from bmq.bmqt cimport PropertyType @@ -154,6 +157,36 @@ cdef class FakeHostHealthMonitor: self._monitor.get().setState(HostHealthState.e_UNHEALTHY) +cdef class FakeAuthnCredentialCb: + cdef object _callback # Store the Python callable + + def __cinit__(self, callback): + self._callback = callback + + # This method will be called by C++ code via PyObject_CallMethod + # Returns None for no credential, or (mechanism, data) tuple + def get_credential_data(self): + try: + result = self._callback() + if result is None: + return None + + if not isinstance(result, tuple) or len(result) != 2: + raise ValueError("callback must return (str, bytes) or None") + + mechanism, data = result + if not isinstance(mechanism, str) or not isinstance(data, bytes): + raise ValueError("callback must return (str, bytes) or None") + + # Return as-is, let C++ side handle conversion + return result + + except Exception: + # Log error or handle as needed + LOGGER.exception("Error in authentication credential callback") + return None + + cdef class Session: cdef object __weakref__ cdef NativeSession* _session @@ -175,6 +208,7 @@ cdef class Session: timeouts: _timeouts.Timeouts = _timeouts.Timeouts(), monitor_host_health: bool = False, fake_host_health_monitor: FakeHostHealthMonitor = None, + fake_authn_credential_cb: FakeAuthnCredentialCb = None, _mock: Optional[object] = None, user_agent_prefix: bytes = b"", ) -> None: @@ -249,6 +283,7 @@ cdef class Session: session_cb, message_cb, ack_cb, + fake_authn_credential_cb, config, fake_host_health_monitor_sp, Error, diff --git a/src/blazingmq/_session.py b/src/blazingmq/_session.py index c7e31f9..225141d 100644 --- a/src/blazingmq/_session.py +++ b/src/blazingmq/_session.py @@ -37,6 +37,7 @@ from ._messages import Message from ._messages import MessageHandle from ._monitors import BasicHealthMonitor +from ._ext import FakeAuthnCredentialCb from ._timeouts import Timeouts from ._typing import PropertyTypeDict from ._typing import PropertyValueDict @@ -302,6 +303,11 @@ class SessionOptions: healthy, `.HostUnhealthy` and `.HostHealthRestored` events with never be emitted, and the *suspends_on_bad_host_health* option of `QueueOptions` cannot be used. + authn_credential_provider: + An optional callable that returns authentication credentials as a + ``(mechanism, data)`` tuple of ``(str, bytes)``, or ``None`` if no + credentials are available. If not provided, no authentication + credentials are sent to the broker. num_processing_threads: The number of threads for the SDK to use for processing events. This defaults to 1. @@ -338,6 +344,7 @@ def __init__( message_compression_algorithm: Optional[CompressionAlgorithmType] = None, timeouts: Optional[Timeouts] = None, host_health_monitor: Union[BasicHealthMonitor, None] = (DefaultMonitor()), + authn_credential_provider: Optional[Callable] = (DefaultAuthnCredentialProvider()), num_processing_threads: Optional[int] = None, blob_buffer_size: Optional[int] = None, channel_high_watermark: Optional[int] = None, @@ -348,6 +355,7 @@ def __init__( self.message_compression_algorithm = message_compression_algorithm self.timeouts = timeouts self.host_health_monitor = host_health_monitor + self.authn_credential_provider = authn_credential_provider self.num_processing_threads = num_processing_threads self.blob_buffer_size = blob_buffer_size self.channel_high_watermark = channel_high_watermark @@ -362,6 +370,7 @@ def __eq__(self, other: object) -> bool: self.message_compression_algorithm == other.message_compression_algorithm and self.timeouts == other.timeouts and self.host_health_monitor == other.host_health_monitor + and self.authn_credential_provider == other.authn_credential_provider and self.num_processing_threads == other.num_processing_threads and self.blob_buffer_size == other.blob_buffer_size and self.channel_high_watermark == other.channel_high_watermark @@ -378,6 +387,7 @@ def __repr__(self) -> str: "message_compression_algorithm", "timeouts", "host_health_monitor", + "authn_credential_provider", "num_processing_threads", "blob_buffer_size", "channel_high_watermark", @@ -426,6 +436,10 @@ class Session: `.HostHealthRestored` events will never be emitted, and the *suspends_on_bad_host_health* option of `QueueOptions` cannot be used. + authn_credential_provider: an optional callable that returns authentication + credentials as a ``(mechanism, data)`` tuple of ``(str, bytes)``, + or ``None`` if no credentials are available. If not provided, no + authentication credentials are sent to the broker. num_processing_threads: The number of threads for the SDK to use for processing events. This defaults to 1. blob_buffer_size: The size (in bytes) of the blob buffers to use. This @@ -470,6 +484,7 @@ def __init__( ), timeout: Union[Timeouts, float] = DEFAULT_TIMEOUT, host_health_monitor: Union[BasicHealthMonitor, None] = (DefaultMonitor()), + authn_credential_provider: Optional[Callable] = None, num_processing_threads: Optional[int] = None, blob_buffer_size: Optional[int] = None, channel_high_watermark: Optional[int] = None, @@ -486,6 +501,11 @@ def __init__( monitor_host_health = host_health_monitor is not None fake_host_health_monitor = getattr(host_health_monitor, "_monitor", None) + fake_authn_credential_provider = ( + FakeAuthnCredentialCb(authn_credential_provider) + if authn_credential_provider is not None + else None + ) self._has_no_on_message = on_message is None @@ -515,6 +535,7 @@ def __init__( timeouts=_validate_timeouts(timeout), monitor_host_health=monitor_host_health, fake_host_health_monitor=fake_host_health_monitor, + fake_authn_credential_cb=fake_authn_credential_provider, user_agent_prefix=_make_user_agent_prefix(user_agent_prefix), ) self._ext.set_owned_by_session() @@ -571,6 +592,7 @@ def with_options( message_compression_algorithm=message_compression_algorithm, timeout=timeout, host_health_monitor=session_options.host_health_monitor, + authn_credential_provider=session_options.authn_credential_provider, num_processing_threads=session_options.num_processing_threads, blob_buffer_size=session_options.blob_buffer_size, channel_high_watermark=session_options.channel_high_watermark, diff --git a/src/cpp/pybmq_session.cpp b/src/cpp/pybmq_session.cpp index 7033987..b92404e 100644 --- a/src/cpp/pybmq_session.cpp +++ b/src/cpp/pybmq_session.cpp @@ -15,6 +15,7 @@ #include +#include #include #include #include @@ -77,6 +78,7 @@ Session::Session( PyObject* py_session_event_callback, PyObject* py_message_event_callback, PyObject* py_ack_event_callback, + PyObject* fake_authn_credential_cb, const SessionConfig& config, bsl::shared_ptr fake_host_health_monitor_sp, PyObject* error, @@ -107,6 +109,74 @@ Session::Session( } d_message_compression_type = config.message_compression_type; + + AuthnCredentialCb cpp_callback; + bool has_auth_callback = false; + + if (fake_authn_credential_cb != nullptr && fake_authn_credential_cb != Py_None) { + // Increment reference count since we're storing the Python object + Py_INCREF(fake_authn_credential_cb); + has_auth_callback = true; + + // Create a C++ lambda that wraps the Python callback + cpp_callback = + [fake_authn_credential_cb]( + bsl::ostream& error) -> bsl::optional { + pybmq::GilAcquireGuard guard; + + // Call get_credential_data() method on the Python object + bslma::ManagedPtr result = + RefUtils::toManagedPtr(PyObject_CallMethod( + fake_authn_credential_cb, + "get_credential_data", + nullptr)); + + if (!result) { + // Python exception occurred + PyErr_Print(); + error << "Error calling get_credential_data()"; + return bsl::optional(); + } + + if (result.get() == Py_None) { + return bsl::optional(); + } + + // Extract tuple (mechanism, data) + if (!PyTuple_Check(result.get()) || PyTuple_Size(result.get()) != 2) { + error << "get_credential_data() must return (str, bytes) or None"; + return bsl::optional(); + } + + PyObject* mechanism_obj = PyTuple_GetItem(result.get(), 0); + PyObject* data_obj = PyTuple_GetItem(result.get(), 1); + + if (!PyUnicode_Check(mechanism_obj) || !PyBytes_Check(data_obj)) { + error << "get_credential_data() must return (str, bytes) or None"; + return bsl::optional(); + } + + // Convert Python str to C++ string + const char* mechanism_cstr = PyUnicode_AsUTF8(mechanism_obj); + bsl::string mechanism(mechanism_cstr); + + // Convert Python bytes to vector + char* data_ptr; + Py_ssize_t data_len; + PyBytes_AsStringAndSize(data_obj, &data_ptr, &data_len); + bsl::vector data(data_ptr, data_ptr + data_len); + + // Construct and return AuthnCredential + bmqt::AuthnCredential credential; + credential.setMechanism(mechanism).setData(data); + + // Move credential into optional (AuthnCredential is move-only) + bsl::optional opt_credential; + opt_credential.emplace(bslmf::MovableRefUtil::move(credential)); + return opt_credential; + }; + } + { pybmq::GilReleaseGuard guard; bmqt::SessionOptions options; @@ -132,6 +202,11 @@ Session::Session( config.event_queue_watermarks.value().second); } + if (has_auth_callback) { + // TODO: This will only compile with setAuthnCredentialCb in SessionOptions + options.setAuthnCredentialCb(cpp_callback); + } + if (config.stats_dump_interval != bsls::TimeInterval()) { options.setStatsDumpInterval(config.stats_dump_interval); } @@ -519,8 +594,8 @@ Session::post( oss << "Failed to post message to " << queue_uri << " queue: " << post_rc; throw GenericError(oss.str()); } - // We have a successful post and the SDK now owns the `on_ack` callback object - // so release our reference without a DECREF. + // We have a successful post and the SDK now owns the `on_ack` callback + // object so release our reference without a DECREF. managed_on_ack.release(); } catch (const GenericError& exc) { PyErr_SetString(d_error, exc.what()); diff --git a/src/cpp/pybmq_session.h b/src/cpp/pybmq_session.h index 8be2b56..6243635 100644 --- a/src/cpp/pybmq_session.h +++ b/src/cpp/pybmq_session.h @@ -23,6 +23,7 @@ #include #include +#include #include #include @@ -49,10 +50,15 @@ class Session Session(const Session&); Session& operator=(const Session&); + // TODO: Remove this once it's added in SessionOptions + typedef bsl::function(bsl::ostream& error)> + AuthnCredentialCb; + public: Session(PyObject* py_session_event_callback, PyObject* py_message_event_callback, PyObject* py_ack_event_callback, + PyObject* fake_authn_credential_cb, const SessionConfig& config, bsl::shared_ptr fake_host_health_monitor, PyObject* d_error, diff --git a/src/declarations/bmq/bmqt.pxd b/src/declarations/bmq/bmqt.pxd index 07e27e9..01b5bf7 100644 --- a/src/declarations/bmq/bmqt.pxd +++ b/src/declarations/bmq/bmqt.pxd @@ -14,6 +14,8 @@ # limitations under the License. from libcpp cimport bool +from bsl cimport string +from bsl cimport vector cdef extern from "bmqt_sessioneventtype.h" namespace "BloombergLP::bmqt::SessionEventType" nogil: @@ -73,3 +75,11 @@ cdef extern from "bmqt_queueoptions.h" namespace "BloombergLP::bmqt::QueueOption int k_DEFAULT_MAX_UNCONFIRMED_BYTES int k_DEFAULT_CONSUMER_PRIORITY bool k_DEFAULT_SUSPENDS_ON_BAD_HOST_HEALTH + +cdef extern from "bmqt_authncredential.h" namespace "BloombergLP::bmqt" nogil: + cdef cppclass AuthnCredential: + AuthnCredential() except + + AuthnCredential& setMechanism(const string&) except + + AuthnCredential& setData(const vector[char]&) except + + const string& mechanism() const + const vector[char]& data() const diff --git a/src/declarations/pybmq.pxd b/src/declarations/pybmq.pxd index 7d21f96..28a18ef 100644 --- a/src/declarations/pybmq.pxd +++ b/src/declarations/pybmq.pxd @@ -58,6 +58,7 @@ cdef extern from "pybmq_session.h" namespace "BloombergLP::pybmq" nogil: Session(object on_session_event, object on_message_event, object on_ack_event, + object fake_authn_credential_cb, const SessionConfig& config, shared_ptr[ManualHostHealthMonitor] fake_host_health_monitor_sp, object error, From d43798199e58300b696b7e2506e7b121ffa93dc9 Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Mon, 22 Jun 2026 15:49:20 -0400 Subject: [PATCH 03/22] picking this up again Signed-off-by: Patrick M. Niedzielski --- src/cpp/pybmq_session.cpp | 2 +- src/cpp/pybmq_session.h | 4 ---- 2 files changed, 1 insertion(+), 5 deletions(-) diff --git a/src/cpp/pybmq_session.cpp b/src/cpp/pybmq_session.cpp index b92404e..14f0fb5 100644 --- a/src/cpp/pybmq_session.cpp +++ b/src/cpp/pybmq_session.cpp @@ -119,6 +119,7 @@ Session::Session( has_auth_callback = true; // Create a C++ lambda that wraps the Python callback + // TODO this can't be a lambda cpp_callback = [fake_authn_credential_cb]( bsl::ostream& error) -> bsl::optional { @@ -203,7 +204,6 @@ Session::Session( } if (has_auth_callback) { - // TODO: This will only compile with setAuthnCredentialCb in SessionOptions options.setAuthnCredentialCb(cpp_callback); } diff --git a/src/cpp/pybmq_session.h b/src/cpp/pybmq_session.h index 6243635..8622558 100644 --- a/src/cpp/pybmq_session.h +++ b/src/cpp/pybmq_session.h @@ -50,10 +50,6 @@ class Session Session(const Session&); Session& operator=(const Session&); - // TODO: Remove this once it's added in SessionOptions - typedef bsl::function(bsl::ostream& error)> - AuthnCredentialCb; - public: Session(PyObject* py_session_event_callback, PyObject* py_message_event_callback, From 7d9303d7dfc265fdfd7312188b1141c1f09a91f8 Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Tue, 23 Jun 2026 17:51:26 -0400 Subject: [PATCH 04/22] Provide `DefaultAuthnCredentialCb` Right now, this defaults to `None` (i.e, no authentication). Signed-off-by: Patrick M. Niedzielski --- src/blazingmq/_session.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/src/blazingmq/_session.py b/src/blazingmq/_session.py index 225141d..0a24d39 100644 --- a/src/blazingmq/_session.py +++ b/src/blazingmq/_session.py @@ -55,6 +55,10 @@ def DefaultMonitor() -> Union[BasicHealthMonitor, None]: return None +def DefaultAuthnCredentialCb() -> Optional[Callable]: + return None + + DEFAULT_TIMEOUT = DefaultTimeoutType() KNOWN_MONITORS = ("blazingmq.BasicHealthMonitor",) @@ -484,7 +488,7 @@ def __init__( ), timeout: Union[Timeouts, float] = DEFAULT_TIMEOUT, host_health_monitor: Union[BasicHealthMonitor, None] = (DefaultMonitor()), - authn_credential_provider: Optional[Callable] = None, + authn_credential_provider: Optional[Callable] = (DefaultAuthnCredentialCb()), num_processing_threads: Optional[int] = None, blob_buffer_size: Optional[int] = None, channel_high_watermark: Optional[int] = None, From 721b7621ee38c47d7205125af441841c8327e0ea Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Fri, 26 Jun 2026 12:13:38 -0400 Subject: [PATCH 05/22] Fix: Compile error from AuthnCredential API Signed-off-by: Patrick M. Niedzielski --- src/cpp/pybmq_session.cpp | 12 +++++------- 1 file changed, 5 insertions(+), 7 deletions(-) diff --git a/src/cpp/pybmq_session.cpp b/src/cpp/pybmq_session.cpp index 14f0fb5..0752646 100644 --- a/src/cpp/pybmq_session.cpp +++ b/src/cpp/pybmq_session.cpp @@ -167,13 +167,11 @@ Session::Session( PyBytes_AsStringAndSize(data_obj, &data_ptr, &data_len); bsl::vector data(data_ptr, data_ptr + data_len); - // Construct and return AuthnCredential - bmqt::AuthnCredential credential; - credential.setMechanism(mechanism).setData(data); - - // Move credential into optional (AuthnCredential is move-only) - bsl::optional opt_credential; - opt_credential.emplace(bslmf::MovableRefUtil::move(credential)); + // Construct and move credential into optional + // (AuthnCredential is move-only) + bmqt::AuthnCredential credential(mechanism, data); + bsl::optional opt_credential( + bslmf::MovableRefUtil::move(credential)); return opt_credential; }; } From 5c9a43aedf3e2811e7c7890ccb12ed4d9c51a0a2 Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Fri, 26 Jun 2026 12:20:05 -0400 Subject: [PATCH 06/22] Fix: Fully qualify `AuthnCredentialCb` Signed-off-by: Patrick M. Niedzielski --- src/cpp/pybmq_session.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/cpp/pybmq_session.cpp b/src/cpp/pybmq_session.cpp index 0752646..4e098d3 100644 --- a/src/cpp/pybmq_session.cpp +++ b/src/cpp/pybmq_session.cpp @@ -110,7 +110,7 @@ Session::Session( d_message_compression_type = config.message_compression_type; - AuthnCredentialCb cpp_callback; + bmqt::SessionOptions::AuthnCredentialCb cpp_callback; bool has_auth_callback = false; if (fake_authn_credential_cb != nullptr && fake_authn_credential_cb != Py_None) { From c53d0286758ec70c29693e2effdf2df1d92ab589 Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Fri, 26 Jun 2026 15:41:21 -0400 Subject: [PATCH 07/22] Fix: Add `fake_authn_credential_cb` value to failing tests Signed-off-by: Patrick M. Niedzielski --- tests/unit/test_session.py | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/tests/unit/test_session.py b/tests/unit/test_session.py index be61bb7..a84e852 100644 --- a/tests/unit/test_session.py +++ b/tests/unit/test_session.py @@ -78,6 +78,7 @@ def dummy2(): ), monitor_host_health=False, fake_host_health_monitor=None, + fake_authn_credential_cb=None, user_agent_prefix=mock.ANY, # varies by version; see dedicated tests ) @@ -129,6 +130,7 @@ def dummy2(): timeouts=timeouts, monitor_host_health=False, fake_host_health_monitor=None, + fake_authn_credential_cb=None, user_agent_prefix=mock.ANY, # varies by version; see dedicated tests ) @@ -174,6 +176,7 @@ def dummy2(): timeouts=timeouts, monitor_host_health=False, fake_host_health_monitor=None, + fake_authn_credential_cb=None, user_agent_prefix=mock.ANY, # varies by version; see dedicated tests ) @@ -210,6 +213,7 @@ def dummy2(): timeouts=Timeouts(), monitor_host_health=False, fake_host_health_monitor=None, + fake_authn_credential_cb=None, user_agent_prefix=mock.ANY, # varies by version; see dedicated tests ) @@ -263,6 +267,7 @@ def dummy2(): timeouts=timeouts, monitor_host_health=False, fake_host_health_monitor=None, + fake_authn_credential_cb=None, user_agent_prefix=mock.ANY, # varies by version; see dedicated tests ) @@ -337,6 +342,7 @@ def dummy2(): ), monitor_host_health=True, fake_host_health_monitor=monitor._monitor, + fake_authn_credential_cb=None, user_agent_prefix=mock.ANY, # varies by version; see dedicated tests ) @@ -369,6 +375,7 @@ def dummy2(): timeouts=Timeouts(), monitor_host_health=False, fake_host_health_monitor=None, + fake_authn_credential_cb=None, user_agent_prefix=mock.ANY, # varies by version; see dedicated tests ) From c98534aa025da2ee15dcb949e6115e0559ebc641 Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Fri, 26 Jun 2026 15:45:09 -0400 Subject: [PATCH 08/22] Fix: Test `authn_credential_provider` in `SessionOptions` Signed-off-by: Patrick M. Niedzielski --- tests/unit/test_session_options.py | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/tests/unit/test_session_options.py b/tests/unit/test_session_options.py index ce14459..43f4026 100644 --- a/tests/unit/test_session_options.py +++ b/tests/unit/test_session_options.py @@ -60,6 +60,7 @@ def test_session_options_default_to_none(): assert options.message_compression_algorithm is None assert options.timeouts is None assert options.host_health_monitor is None + assert options.authn_credential_provider is None assert options.num_processing_threads is None assert options.blob_buffer_size is None assert options.channel_high_watermark is None @@ -95,6 +96,7 @@ def test_session_options_equality(): blazingmq.SessionOptions(channel_high_watermark=8000000), blazingmq.SessionOptions(event_queue_watermarks=(6000000, 7000000)), blazingmq.SessionOptions(stats_dump_interval=30.0), + blazingmq.SessionOptions(authn_credential_provider=lambda: None), blazingmq.SessionOptions(user_agent_prefix=b"mylib:1.0"), ], ) @@ -104,3 +106,15 @@ def test_queue_options_other_inequality(right): # THEN assert not left == right + + +def test_session_options_repr_with_authn_credential_provider(): + # GIVEN + def my_provider(): + return ("mechanism", b"data") + + # WHEN + options = blazingmq.SessionOptions(authn_credential_provider=my_provider) + + # THEN + assert "authn_credential_provider=" in repr(options) From 04e7876950d819ad522b90c57f1a25fd0b16b576 Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Fri, 26 Jun 2026 15:54:03 -0400 Subject: [PATCH 09/22] Test: Add tests for `ExtSession` construction Signed-off-by: Patrick M. Niedzielski --- tests/unit/test_session.py | 143 +++++++++++++++++++++++++++++++++++++ 1 file changed, 143 insertions(+) diff --git a/tests/unit/test_session.py b/tests/unit/test_session.py index a84e852..1df5bb8 100644 --- a/tests/unit/test_session.py +++ b/tests/unit/test_session.py @@ -83,6 +83,56 @@ def dummy2(): ) +@mock.patch("blazingmq._session.ExtSession") +def test_session_constructed_with_authn_credential_provider(ext_cls): + # GIVEN + ext_cls.mock_add_spec([]) + + def dummy1(): + pass + + def dummy2(): + pass + + def my_provider(): + return ("mechanism", b"data") + + # WHEN + Session( + dummy1, + on_message=dummy2, + broker="some_uri", + timeout=60.0, + host_health_monitor=None, + authn_credential_provider=my_provider, + ) + + # THEN + ext_cls.assert_called_once_with( + dummy1, + on_message=dummy2, + broker=b"some_uri", + message_compression_algorithm=CompressionAlgorithmType.NONE, + num_processing_threads=None, + blob_buffer_size=None, + channel_high_watermark=None, + event_queue_watermarks=None, + stats_dump_interval=None, + timeouts=Timeouts( + connect_timeout=None, + disconnect_timeout=None, + open_queue_timeout=60.0, + configure_queue_timeout=60.0, + close_queue_timeout=60.0, + ), + monitor_host_health=False, + fake_host_health_monitor=None, + fake_authn_credential_cb=mock.ANY, + ) + call_kwargs = ext_cls.call_args[1] + assert call_kwargs["fake_authn_credential_cb"] is not None + + @mock.patch("blazingmq._session.ExtSession") def test_session_constructed_with_timeouts(ext_cls): # GIVEN @@ -300,6 +350,99 @@ def dummy2(): ) +@mock.patch("blazingmq._session.ExtSession") +def test_session_default_with_options_authn_credential_provider(ext_cls): + # GIVEN + ext_cls.mock_add_spec([]) + + def dummy1(): + pass + + def dummy2(): + pass + + def my_provider(): + return ("mechanism", b"data") + + session_options = SessionOptions(authn_credential_provider=my_provider) + + # WHEN + Session.with_options( + dummy1, on_message=dummy2, broker="some_uri", session_options=session_options + ) + + # THEN + ext_cls.assert_called_once_with( + dummy1, + on_message=dummy2, + broker=b"some_uri", + message_compression_algorithm=CompressionAlgorithmType.NONE, + num_processing_threads=None, + blob_buffer_size=None, + channel_high_watermark=None, + event_queue_watermarks=None, + stats_dump_interval=None, + timeouts=Timeouts(), + monitor_host_health=False, + fake_host_health_monitor=None, + fake_authn_credential_cb=mock.ANY, + ) + call_kwargs = ext_cls.call_args[1] + assert call_kwargs["fake_authn_credential_cb"] is not None + + +@mock.patch("blazingmq._session.ExtSession") +def test_session_with_options_authn_credential_provider(ext_cls): + # GIVEN + ext_cls.mock_add_spec([]) + + def dummy1(): + pass + + def dummy2(): + pass + + def my_provider(): + return ("mechanism", b"data") + + timeouts = Timeouts( + connect_timeout=60.0, + disconnect_timeout=70.0, + open_queue_timeout=80.0, + configure_queue_timeout=90.0, + close_queue_timeout=100.0, + ) + + session_options = SessionOptions( + timeouts=timeouts, + authn_credential_provider=my_provider, + ) + + # WHEN + Session.with_options( + dummy1, on_message=dummy2, broker="some_uri", session_options=session_options + ) + + # THEN + ext_cls.assert_called_once_with( + dummy1, + on_message=dummy2, + broker=b"some_uri", + message_compression_algorithm=CompressionAlgorithmType.NONE, + num_processing_threads=None, + blob_buffer_size=None, + channel_high_watermark=None, + event_queue_watermarks=None, + stats_dump_interval=None, + timeouts=timeouts, + monitor_host_health=False, + fake_host_health_monitor=None, + fake_authn_credential_cb=mock.ANY, + ) + call_kwargs = ext_cls.call_args[1] + assert call_kwargs["fake_authn_credential_cb"] is not None + + @mock.patch("blazingmq._session.ExtSession") def test_session_basic_monitor(ext_cls): # GIVEN From 38e111fc60651b69aa9f2c94fe640236eb875805 Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Fri, 26 Jun 2026 15:55:11 -0400 Subject: [PATCH 10/22] Fix: Format `DefaultAuthnCredentialProvider` Signed-off-by: Patrick M. Niedzielski --- src/blazingmq/_session.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/src/blazingmq/_session.py b/src/blazingmq/_session.py index 0a24d39..1300868 100644 --- a/src/blazingmq/_session.py +++ b/src/blazingmq/_session.py @@ -348,7 +348,9 @@ def __init__( message_compression_algorithm: Optional[CompressionAlgorithmType] = None, timeouts: Optional[Timeouts] = None, host_health_monitor: Union[BasicHealthMonitor, None] = (DefaultMonitor()), - authn_credential_provider: Optional[Callable] = (DefaultAuthnCredentialProvider()), + authn_credential_provider: Optional[Callable] = ( + DefaultAuthnCredentialProvider() + ), num_processing_threads: Optional[int] = None, blob_buffer_size: Optional[int] = None, channel_high_watermark: Optional[int] = None, @@ -488,7 +490,9 @@ def __init__( ), timeout: Union[Timeouts, float] = DEFAULT_TIMEOUT, host_health_monitor: Union[BasicHealthMonitor, None] = (DefaultMonitor()), - authn_credential_provider: Optional[Callable] = (DefaultAuthnCredentialCb()), + authn_credential_provider: Optional[Callable] = ( + DefaultAuthnCredentialCb() + ), num_processing_threads: Optional[int] = None, blob_buffer_size: Optional[int] = None, channel_high_watermark: Optional[int] = None, From c58f5f54601b9178d623b1d8c035b0b2e4b40cdd Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Fri, 26 Jun 2026 16:09:53 -0400 Subject: [PATCH 11/22] Fix: Update `AuthnCredential` in bmqt.pxd Signed-off-by: Patrick M. Niedzielski --- src/declarations/bmq/bmqt.pxd | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/src/declarations/bmq/bmqt.pxd b/src/declarations/bmq/bmqt.pxd index 01b5bf7..5f05d12 100644 --- a/src/declarations/bmq/bmqt.pxd +++ b/src/declarations/bmq/bmqt.pxd @@ -1,4 +1,4 @@ -# Copyright 2019-2023 Bloomberg Finance L.P. +# Copyright 2019-2026 Bloomberg Finance L.P. # SPDX-License-Identifier: Apache-2.0 # # Licensed under the Apache License, Version 2.0 (the "License"); @@ -13,9 +13,9 @@ # See the License for the specific language governing permissions and # limitations under the License. -from libcpp cimport bool from bsl cimport string from bsl cimport vector +from libcpp cimport bool cdef extern from "bmqt_sessioneventtype.h" namespace "BloombergLP::bmqt::SessionEventType" nogil: @@ -79,7 +79,7 @@ cdef extern from "bmqt_queueoptions.h" namespace "BloombergLP::bmqt::QueueOption cdef extern from "bmqt_authncredential.h" namespace "BloombergLP::bmqt" nogil: cdef cppclass AuthnCredential: AuthnCredential() except + - AuthnCredential& setMechanism(const string&) except + - AuthnCredential& setData(const vector[char]&) except + + AuthnCredential(const AuthnCredential&) except + + AuthnCredential(const string& mechanism, const vector[char]& data) except + const string& mechanism() const const vector[char]& data() const From c1b400e0855ab5e138f199a1a1b4d54bd7d579d1 Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Tue, 30 Jun 2026 14:37:26 -0400 Subject: [PATCH 12/22] Rename `FakeAuthnCredentialCb` Signed-off-by: Patrick M. Niedzielski --- src/blazingmq/_ext.pyi | 4 +- src/blazingmq/_ext.pyx | 6 +- src/blazingmq/_session.py | 8 +- src/cpp/pybmq_session.cpp | 10 +- src/cpp/pybmq_session.h | 2 +- src/declarations/pybmq.pxd | 2 +- .../unit/test_authn_credential_cb_adapter.py | 114 ++++++++++++++++++ tests/unit/test_session.py | 26 ++-- 8 files changed, 143 insertions(+), 29 deletions(-) create mode 100644 tests/unit/test_authn_credential_cb_adapter.py diff --git a/src/blazingmq/_ext.pyi b/src/blazingmq/_ext.pyi index de49aa7..7b94abc 100644 --- a/src/blazingmq/_ext.pyi +++ b/src/blazingmq/_ext.pyi @@ -37,7 +37,7 @@ class FakeHostHealthMonitor: def set_healthy(self) -> None: ... def set_unhealthy(self) -> None: ... -class FakeAuthnCredentialCb: +class AuthnCredentialCbAdapter: def __init__(self, callback: Callable[[], Optional[tuple[str, bytes]]]) -> None: ... class Session: @@ -56,7 +56,7 @@ class Session: timeouts: Timeouts = Timeouts(), monitor_host_health: bool = False, fake_host_health_monitor: Optional[FakeHostHealthMonitor] = None, - fake_authn_credential_cb: Optional[FakeAuthnCredentialCb] = None, + authn_credential_cb: Optional[AuthnCredentialCbAdapter] = None, user_agent_prefix: bytes = b"", ) -> None: ... def stop(self) -> None: ... diff --git a/src/blazingmq/_ext.pyx b/src/blazingmq/_ext.pyx index f1b0d8e..430cb24 100644 --- a/src/blazingmq/_ext.pyx +++ b/src/blazingmq/_ext.pyx @@ -157,7 +157,7 @@ cdef class FakeHostHealthMonitor: self._monitor.get().setState(HostHealthState.e_UNHEALTHY) -cdef class FakeAuthnCredentialCb: +cdef class AuthnCredentialCbAdapter: cdef object _callback # Store the Python callable def __cinit__(self, callback): @@ -208,7 +208,7 @@ cdef class Session: timeouts: _timeouts.Timeouts = _timeouts.Timeouts(), monitor_host_health: bool = False, fake_host_health_monitor: FakeHostHealthMonitor = None, - fake_authn_credential_cb: FakeAuthnCredentialCb = None, + authn_credential_cb: AuthnCredentialCbAdapter = None, _mock: Optional[object] = None, user_agent_prefix: bytes = b"", ) -> None: @@ -283,7 +283,7 @@ cdef class Session: session_cb, message_cb, ack_cb, - fake_authn_credential_cb, + authn_credential_cb, config, fake_host_health_monitor_sp, Error, diff --git a/src/blazingmq/_session.py b/src/blazingmq/_session.py index 1300868..9bce84e 100644 --- a/src/blazingmq/_session.py +++ b/src/blazingmq/_session.py @@ -27,6 +27,7 @@ from ._about import __version__ from ._enums import CompressionAlgorithmType from ._enums import PropertyType +from ._ext import AuthnCredentialCbAdapter from ._ext import DEFAULT_CONSUMER_PRIORITY from ._ext import DEFAULT_MAX_UNCONFIRMED_BYTES from ._ext import DEFAULT_MAX_UNCONFIRMED_MESSAGES @@ -37,7 +38,6 @@ from ._messages import Message from ._messages import MessageHandle from ._monitors import BasicHealthMonitor -from ._ext import FakeAuthnCredentialCb from ._timeouts import Timeouts from ._typing import PropertyTypeDict from ._typing import PropertyValueDict @@ -509,8 +509,8 @@ def __init__( monitor_host_health = host_health_monitor is not None fake_host_health_monitor = getattr(host_health_monitor, "_monitor", None) - fake_authn_credential_provider = ( - FakeAuthnCredentialCb(authn_credential_provider) + authn_credential_cb = ( + AuthnCredentialCbAdapter(authn_credential_provider) if authn_credential_provider is not None else None ) @@ -543,7 +543,7 @@ def __init__( timeouts=_validate_timeouts(timeout), monitor_host_health=monitor_host_health, fake_host_health_monitor=fake_host_health_monitor, - fake_authn_credential_cb=fake_authn_credential_provider, + authn_credential_cb=authn_credential_cb, user_agent_prefix=_make_user_agent_prefix(user_agent_prefix), ) self._ext.set_owned_by_session() diff --git a/src/cpp/pybmq_session.cpp b/src/cpp/pybmq_session.cpp index 4e098d3..d94fc75 100644 --- a/src/cpp/pybmq_session.cpp +++ b/src/cpp/pybmq_session.cpp @@ -78,7 +78,7 @@ Session::Session( PyObject* py_session_event_callback, PyObject* py_message_event_callback, PyObject* py_ack_event_callback, - PyObject* fake_authn_credential_cb, + PyObject* authn_credential_cb, const SessionConfig& config, bsl::shared_ptr fake_host_health_monitor_sp, PyObject* error, @@ -113,22 +113,22 @@ Session::Session( bmqt::SessionOptions::AuthnCredentialCb cpp_callback; bool has_auth_callback = false; - if (fake_authn_credential_cb != nullptr && fake_authn_credential_cb != Py_None) { + if (authn_credential_cb != nullptr && authn_credential_cb != Py_None) { // Increment reference count since we're storing the Python object - Py_INCREF(fake_authn_credential_cb); + Py_INCREF(authn_credential_cb); has_auth_callback = true; // Create a C++ lambda that wraps the Python callback // TODO this can't be a lambda cpp_callback = - [fake_authn_credential_cb]( + [authn_credential_cb]( bsl::ostream& error) -> bsl::optional { pybmq::GilAcquireGuard guard; // Call get_credential_data() method on the Python object bslma::ManagedPtr result = RefUtils::toManagedPtr(PyObject_CallMethod( - fake_authn_credential_cb, + authn_credential_cb, "get_credential_data", nullptr)); diff --git a/src/cpp/pybmq_session.h b/src/cpp/pybmq_session.h index 8622558..3a5ff74 100644 --- a/src/cpp/pybmq_session.h +++ b/src/cpp/pybmq_session.h @@ -54,7 +54,7 @@ class Session Session(PyObject* py_session_event_callback, PyObject* py_message_event_callback, PyObject* py_ack_event_callback, - PyObject* fake_authn_credential_cb, + PyObject* authn_credential_cb, const SessionConfig& config, bsl::shared_ptr fake_host_health_monitor, PyObject* d_error, diff --git a/src/declarations/pybmq.pxd b/src/declarations/pybmq.pxd index 28a18ef..d51a45d 100644 --- a/src/declarations/pybmq.pxd +++ b/src/declarations/pybmq.pxd @@ -58,7 +58,7 @@ cdef extern from "pybmq_session.h" namespace "BloombergLP::pybmq" nogil: Session(object on_session_event, object on_message_event, object on_ack_event, - object fake_authn_credential_cb, + object authn_credential_cb, const SessionConfig& config, shared_ptr[ManualHostHealthMonitor] fake_host_health_monitor_sp, object error, diff --git a/tests/unit/test_authn_credential_cb_adapter.py b/tests/unit/test_authn_credential_cb_adapter.py new file mode 100644 index 0000000..4e639e2 --- /dev/null +++ b/tests/unit/test_authn_credential_cb_adapter.py @@ -0,0 +1,114 @@ +# Copyright 2026 Bloomberg Finance L.P. +# SPDX-License-Identifier: Apache-2.0 +# +# 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. + +from blazingmq._ext import AuthnCredentialCbAdapter + + +def test_valid_return(): + # GIVEN + def provider(): + return ("mechanism", b"data") + + adapter = AuthnCredentialCbAdapter(provider) + + # WHEN + result = adapter.get_credential_data() + + # THEN + assert result == ("mechanism", b"data") + + +def test_none_return(): + # GIVEN + def provider(): + return None + + adapter = AuthnCredentialCbAdapter(provider) + + # WHEN + result = adapter.get_credential_data() + + # THEN + assert result is None + + +def test_not_a_tuple(): + # GIVEN + def provider(): + return "not a tuple" + + adapter = AuthnCredentialCbAdapter(provider) + + # WHEN + result = adapter.get_credential_data() + + # THEN + assert result is None + + +def test_tuple_wrong_length(): + # GIVEN + def provider(): + return ("mechanism", b"data", "extra") + + adapter = AuthnCredentialCbAdapter(provider) + + # WHEN + result = adapter.get_credential_data() + + # THEN + assert result is None + + +def test_mechanism_not_str(): + # GIVEN + def provider(): + return (123, b"data") + + adapter = AuthnCredentialCbAdapter(provider) + + # WHEN + result = adapter.get_credential_data() + + # THEN + assert result is None + + +def test_data_not_bytes(): + # GIVEN + def provider(): + return ("mechanism", "not bytes") + + adapter = AuthnCredentialCbAdapter(provider) + + # WHEN + result = adapter.get_credential_data() + + # THEN + assert result is None + + +def test_callback_raises(): + # GIVEN + def provider(): + raise RuntimeError("broken") + + adapter = AuthnCredentialCbAdapter(provider) + + # WHEN + result = adapter.get_credential_data() + + # THEN + assert result is None diff --git a/tests/unit/test_session.py b/tests/unit/test_session.py index 1df5bb8..9b2258f 100644 --- a/tests/unit/test_session.py +++ b/tests/unit/test_session.py @@ -78,7 +78,7 @@ def dummy2(): ), monitor_host_health=False, fake_host_health_monitor=None, - fake_authn_credential_cb=None, + authn_credential_cb=None, user_agent_prefix=mock.ANY, # varies by version; see dedicated tests ) @@ -127,10 +127,10 @@ def my_provider(): ), monitor_host_health=False, fake_host_health_monitor=None, - fake_authn_credential_cb=mock.ANY, + authn_credential_cb=mock.ANY, ) call_kwargs = ext_cls.call_args[1] - assert call_kwargs["fake_authn_credential_cb"] is not None + assert call_kwargs["authn_credential_cb"] is not None @mock.patch("blazingmq._session.ExtSession") @@ -180,7 +180,7 @@ def dummy2(): timeouts=timeouts, monitor_host_health=False, fake_host_health_monitor=None, - fake_authn_credential_cb=None, + authn_credential_cb=None, user_agent_prefix=mock.ANY, # varies by version; see dedicated tests ) @@ -226,7 +226,7 @@ def dummy2(): timeouts=timeouts, monitor_host_health=False, fake_host_health_monitor=None, - fake_authn_credential_cb=None, + authn_credential_cb=None, user_agent_prefix=mock.ANY, # varies by version; see dedicated tests ) @@ -263,7 +263,7 @@ def dummy2(): timeouts=Timeouts(), monitor_host_health=False, fake_host_health_monitor=None, - fake_authn_credential_cb=None, + authn_credential_cb=None, user_agent_prefix=mock.ANY, # varies by version; see dedicated tests ) @@ -317,7 +317,7 @@ def dummy2(): timeouts=timeouts, monitor_host_health=False, fake_host_health_monitor=None, - fake_authn_credential_cb=None, + authn_credential_cb=None, user_agent_prefix=mock.ANY, # varies by version; see dedicated tests ) @@ -385,10 +385,10 @@ def my_provider(): timeouts=Timeouts(), monitor_host_health=False, fake_host_health_monitor=None, - fake_authn_credential_cb=mock.ANY, + authn_credential_cb=mock.ANY, ) call_kwargs = ext_cls.call_args[1] - assert call_kwargs["fake_authn_credential_cb"] is not None + assert call_kwargs["authn_credential_cb"] is not None @mock.patch("blazingmq._session.ExtSession") @@ -437,10 +437,10 @@ def my_provider(): timeouts=timeouts, monitor_host_health=False, fake_host_health_monitor=None, - fake_authn_credential_cb=mock.ANY, + authn_credential_cb=mock.ANY, ) call_kwargs = ext_cls.call_args[1] - assert call_kwargs["fake_authn_credential_cb"] is not None + assert call_kwargs["authn_credential_cb"] is not None @mock.patch("blazingmq._session.ExtSession") @@ -485,7 +485,7 @@ def dummy2(): ), monitor_host_health=True, fake_host_health_monitor=monitor._monitor, - fake_authn_credential_cb=None, + authn_credential_cb=None, user_agent_prefix=mock.ANY, # varies by version; see dedicated tests ) @@ -518,7 +518,7 @@ def dummy2(): timeouts=Timeouts(), monitor_host_health=False, fake_host_health_monitor=None, - fake_authn_credential_cb=None, + authn_credential_cb=None, user_agent_prefix=mock.ANY, # varies by version; see dedicated tests ) From 46585dcb4af64a9320be4d1354a1f74eca743a15 Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Tue, 30 Jun 2026 14:37:42 -0400 Subject: [PATCH 13/22] Fix `isort` order Signed-off-by: Patrick M. Niedzielski --- src/blazingmq/_ext.pyx | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/blazingmq/_ext.pyx b/src/blazingmq/_ext.pyx index 430cb24..1f97b09 100644 --- a/src/blazingmq/_ext.pyx +++ b/src/blazingmq/_ext.pyx @@ -21,8 +21,8 @@ import weakref from bsl cimport optional from bsl cimport pair from bsl cimport shared_ptr -from bsl cimport vector from bsl cimport string +from bsl cimport vector from bsl.bsls cimport TimeInterval from cpython.ceval cimport PyEval_InitThreads from libcpp cimport bool as cppbool From c9d0f7402048421e0aa34928789584fa20c1ff92 Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Tue, 30 Jun 2026 15:19:39 -0400 Subject: [PATCH 14/22] Add `AuthnCredentialProvider` type alias Signed-off-by: Patrick M. Niedzielski --- src/blazingmq/__init__.py | 2 ++ src/blazingmq/_session.py | 9 +++++---- src/blazingmq/_typing.py | 8 ++++++++ 3 files changed, 15 insertions(+), 4 deletions(-) diff --git a/src/blazingmq/__init__.py b/src/blazingmq/__init__.py index 910d1ad..28d81ef 100644 --- a/src/blazingmq/__init__.py +++ b/src/blazingmq/__init__.py @@ -27,6 +27,7 @@ from ._session import Session from ._session import SessionOptions from ._timeouts import Timeouts +from ._typing import AuthnCredentialProvider from ._typing import PropertyTypeDict from ._typing import PropertyValueDict from .exceptions import Error @@ -34,6 +35,7 @@ __all__ = [ "Ack", "AckStatus", + "AuthnCredentialProvider", "BasicHealthMonitor", "CompressionAlgorithmType", "Error", diff --git a/src/blazingmq/_session.py b/src/blazingmq/_session.py index 9bce84e..809b2c1 100644 --- a/src/blazingmq/_session.py +++ b/src/blazingmq/_session.py @@ -39,6 +39,7 @@ from ._messages import MessageHandle from ._monitors import BasicHealthMonitor from ._timeouts import Timeouts +from ._typing import AuthnCredentialProvider from ._typing import PropertyTypeDict from ._typing import PropertyValueDict from ._typing import PropertyValueType @@ -55,7 +56,7 @@ def DefaultMonitor() -> Union[BasicHealthMonitor, None]: return None -def DefaultAuthnCredentialCb() -> Optional[Callable]: +def DefaultAuthnCredentialProvider() -> Optional[AuthnCredentialProvider]: return None @@ -348,7 +349,7 @@ def __init__( message_compression_algorithm: Optional[CompressionAlgorithmType] = None, timeouts: Optional[Timeouts] = None, host_health_monitor: Union[BasicHealthMonitor, None] = (DefaultMonitor()), - authn_credential_provider: Optional[Callable] = ( + authn_credential_provider: Optional[AuthnCredentialProvider] = ( DefaultAuthnCredentialProvider() ), num_processing_threads: Optional[int] = None, @@ -490,8 +491,8 @@ def __init__( ), timeout: Union[Timeouts, float] = DEFAULT_TIMEOUT, host_health_monitor: Union[BasicHealthMonitor, None] = (DefaultMonitor()), - authn_credential_provider: Optional[Callable] = ( - DefaultAuthnCredentialCb() + authn_credential_provider: Optional[AuthnCredentialProvider] = ( + DefaultAuthnCredentialProvider() ), num_processing_threads: Optional[int] = None, blob_buffer_size: Optional[int] = None, diff --git a/src/blazingmq/_typing.py b/src/blazingmq/_typing.py index 39cf633..7c09d27 100644 --- a/src/blazingmq/_typing.py +++ b/src/blazingmq/_typing.py @@ -13,7 +13,9 @@ # See the License for the specific language governing permissions and # limitations under the License. +from typing import Callable from typing import Mapping +from typing import Optional from typing import Union from ._enums import PropertyType @@ -23,3 +25,9 @@ PropertyValueDict = Mapping[str, PropertyValueType] PropertyTypeDict = Mapping[str, PropertyType] + +AuthnCredentialProvider = Callable[[], Optional[tuple[str, bytes]]] +"""A callable that returns authentication credentials as a tuple of +``(mechanism, data)`` tuple of ``(str, bytes)``, or ``None`` if an +error occurs while obtaining credentials. +""" From 97926e9822fa5c942ef57cbc62dee76170997b54 Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Tue, 30 Jun 2026 18:29:04 -0400 Subject: [PATCH 15/22] clang-format C++ code Signed-off-by: Patrick M. Niedzielski --- src/cpp/pybmq_session.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/cpp/pybmq_session.cpp b/src/cpp/pybmq_session.cpp index d94fc75..4dc4a3a 100644 --- a/src/cpp/pybmq_session.cpp +++ b/src/cpp/pybmq_session.cpp @@ -171,7 +171,7 @@ Session::Session( // (AuthnCredential is move-only) bmqt::AuthnCredential credential(mechanism, data); bsl::optional opt_credential( - bslmf::MovableRefUtil::move(credential)); + bslmf::MovableRefUtil::move(credential)); return opt_credential; }; } From ed28229b0c23ba6e70e23538989598e9c03763d3 Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Wed, 1 Jul 2026 11:02:49 -0400 Subject: [PATCH 16/22] Add documentation for `AuthnCredentialProvider` Signed-off-by: Patrick M. Niedzielski --- docs/api_reference.rst | 2 ++ src/blazingmq/_session.py | 11 ++++++----- 2 files changed, 8 insertions(+), 5 deletions(-) diff --git a/docs/api_reference.rst b/docs/api_reference.rst index b380ab8..438ff85 100644 --- a/docs/api_reference.rst +++ b/docs/api_reference.rst @@ -187,6 +187,8 @@ Testing Utilities Helper Types ============ +.. autoclass:: blazingmq.AuthnCredentialProvider + .. autoclass:: blazingmq.PropertyTypeDict .. autoclass:: blazingmq.PropertyValueDict diff --git a/src/blazingmq/_session.py b/src/blazingmq/_session.py index 809b2c1..ff259a2 100644 --- a/src/blazingmq/_session.py +++ b/src/blazingmq/_session.py @@ -308,7 +308,7 @@ class SessionOptions: healthy, `.HostUnhealthy` and `.HostHealthRestored` events with never be emitted, and the *suspends_on_bad_host_health* option of `QueueOptions` cannot be used. - authn_credential_provider: + authn_credential_provider (Optional[`~blazingmq.AuthnCredentialProvider`]): An optional callable that returns authentication credentials as a ``(mechanism, data)`` tuple of ``(str, bytes)``, or ``None`` if no credentials are available. If not provided, no authentication @@ -443,10 +443,11 @@ class Session: `.HostHealthRestored` events will never be emitted, and the *suspends_on_bad_host_health* option of `QueueOptions` cannot be used. - authn_credential_provider: an optional callable that returns authentication - credentials as a ``(mechanism, data)`` tuple of ``(str, bytes)``, - or ``None`` if no credentials are available. If not provided, no - authentication credentials are sent to the broker. + authn_credential_provider (Optional[`~blazingmq.AuthnCredentialProvider`]): + an optional callable that returns authentication credentials as a + ``(mechanism, data)`` tuple of ``(str, bytes)``, or ``None`` if no + credentials are available. If not provided, no authentication + credentials are sent to the broker. num_processing_threads: The number of threads for the SDK to use for processing events. This defaults to 1. blob_buffer_size: The size (in bytes) of the blob buffers to use. This From 306436a268def0cc36846a6661d60281c98c6020 Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Mon, 31 Aug 2026 17:19:19 -0400 Subject: [PATCH 17/22] Fix: Remove `ostream& error` from authn callback Signed-off-by: Patrick M. Niedzielski --- src/cpp/pybmq_session.cpp | 18 +++++++++++------- 1 file changed, 11 insertions(+), 7 deletions(-) diff --git a/src/cpp/pybmq_session.cpp b/src/cpp/pybmq_session.cpp index 4dc4a3a..3156393 100644 --- a/src/cpp/pybmq_session.cpp +++ b/src/cpp/pybmq_session.cpp @@ -22,6 +22,7 @@ #include #include +#include #include #include #include @@ -120,9 +121,9 @@ Session::Session( // Create a C++ lambda that wraps the Python callback // TODO this can't be a lambda - cpp_callback = - [authn_credential_cb]( - bsl::ostream& error) -> bsl::optional { + cpp_callback = [authn_credential_cb]() -> bsl::optional { + BALL_LOG_SET_CATEGORY("pybmq_session"); + pybmq::GilAcquireGuard guard; // Call get_credential_data() method on the Python object @@ -133,9 +134,10 @@ Session::Session( nullptr)); if (!result) { - // Python exception occurred + // Python exception occurred. Clear it before logging, so we + // don't re-enter Python via the BALL observer with it set. PyErr_Print(); - error << "Error calling get_credential_data()"; + BALL_LOG_ERROR << "Error calling get_credential_data()"; return bsl::optional(); } @@ -145,7 +147,8 @@ Session::Session( // Extract tuple (mechanism, data) if (!PyTuple_Check(result.get()) || PyTuple_Size(result.get()) != 2) { - error << "get_credential_data() must return (str, bytes) or None"; + BALL_LOG_ERROR + << "get_credential_data() must return (str, bytes) or None"; return bsl::optional(); } @@ -153,7 +156,8 @@ Session::Session( PyObject* data_obj = PyTuple_GetItem(result.get(), 1); if (!PyUnicode_Check(mechanism_obj) || !PyBytes_Check(data_obj)) { - error << "get_credential_data() must return (str, bytes) or None"; + BALL_LOG_ERROR + << "get_credential_data() must return (str, bytes) or None"; return bsl::optional(); } From f772b39863a16b299e22a8abeaaff6831add348e Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Tue, 1 Sep 2026 10:55:13 -0400 Subject: [PATCH 18/22] Remove lambda for C++03 compat Signed-off-by: Patrick M. Niedzielski --- src/cpp/pybmq_session.cpp | 155 ++++++++++++++++++++++---------------- src/cpp/pybmq_session.h | 1 + 2 files changed, 90 insertions(+), 66 deletions(-) diff --git a/src/cpp/pybmq_session.cpp b/src/cpp/pybmq_session.cpp index 3156393..ceca6ed 100644 --- a/src/cpp/pybmq_session.cpp +++ b/src/cpp/pybmq_session.cpp @@ -73,6 +73,81 @@ class BrokerTimeoutError : public bsl::runtime_error } }; +// Invoke `get_credential_data()` on a Python object and convert the result to +// a `bmqt::AuthnCredential`. The object is held, not owned: the `Session` +// holds the reference and drops it only after destroying the `bmqa::Session` +// that owns every copy of this functor. +class AuthnCredentialCbFunctor +{ + private: + // DATA + PyObject* d_callback_p; + + public: + // CREATORS + explicit AuthnCredentialCbFunctor(PyObject* callback); + + // ACCESSORS + bsl::optional operator()() const; +}; + +AuthnCredentialCbFunctor::AuthnCredentialCbFunctor(PyObject* callback) +: d_callback_p(callback) +{ +} + +bsl::optional +AuthnCredentialCbFunctor::operator()() const +{ + BALL_LOG_SET_CATEGORY("pybmq_session"); + + pybmq::GilAcquireGuard guard; + + // Call get_credential_data() method on the Python object + bslma::ManagedPtr result = RefUtils::toManagedPtr( + PyObject_CallMethod(d_callback_p, "get_credential_data", NULL)); + + if (!result) { + // Python exception occurred. Clear it before logging, so we + // don't re-enter Python via the BALL observer with it set. + PyErr_Print(); + BALL_LOG_ERROR << "Error calling get_credential_data()"; + return bsl::optional(); + } + + if (result.get() == Py_None) { + return bsl::optional(); + } + + // Extract tuple (mechanism, data) + if (!PyTuple_Check(result.get()) || PyTuple_Size(result.get()) != 2) { + BALL_LOG_ERROR << "get_credential_data() must return (str, bytes) or None"; + return bsl::optional(); + } + + PyObject* mechanism_obj = PyTuple_GetItem(result.get(), 0); + PyObject* data_obj = PyTuple_GetItem(result.get(), 1); + + if (!PyUnicode_Check(mechanism_obj) || !PyBytes_Check(data_obj)) { + BALL_LOG_ERROR << "get_credential_data() must return (str, bytes) or None"; + return bsl::optional(); + } + + // Convert Python str to C++ string + const char* mechanism_cstr = PyUnicode_AsUTF8(mechanism_obj); + bsl::string mechanism(mechanism_cstr); + + // Convert Python bytes to vector + char* data_ptr; + Py_ssize_t data_len; + PyBytes_AsStringAndSize(data_obj, &data_ptr, &data_len); + bsl::vector data(data_ptr, data_ptr + data_len); + + bmqt::AuthnCredential credential(mechanism, data); + return bsl::optional( + bslmf::MovableRefUtil::move(credential)); +} + } // namespace Session::Session( @@ -91,6 +166,7 @@ Session::Session( , d_message_compression_type(bmqt::CompressionAlgorithmType::e_NONE) , d_error(error) , d_broker_timeout_error(broker_timeout_error) +, d_authn_credential_cb(NULL) , d_session_mp() { bsl::shared_ptr host_health_monitor_sp; @@ -114,70 +190,10 @@ Session::Session( bmqt::SessionOptions::AuthnCredentialCb cpp_callback; bool has_auth_callback = false; - if (authn_credential_cb != nullptr && authn_credential_cb != Py_None) { - // Increment reference count since we're storing the Python object - Py_INCREF(authn_credential_cb); + if (authn_credential_cb != NULL && authn_credential_cb != Py_None) { + d_authn_credential_cb = authn_credential_cb; + cpp_callback = AuthnCredentialCbFunctor(d_authn_credential_cb); has_auth_callback = true; - - // Create a C++ lambda that wraps the Python callback - // TODO this can't be a lambda - cpp_callback = [authn_credential_cb]() -> bsl::optional { - BALL_LOG_SET_CATEGORY("pybmq_session"); - - pybmq::GilAcquireGuard guard; - - // Call get_credential_data() method on the Python object - bslma::ManagedPtr result = - RefUtils::toManagedPtr(PyObject_CallMethod( - authn_credential_cb, - "get_credential_data", - nullptr)); - - if (!result) { - // Python exception occurred. Clear it before logging, so we - // don't re-enter Python via the BALL observer with it set. - PyErr_Print(); - BALL_LOG_ERROR << "Error calling get_credential_data()"; - return bsl::optional(); - } - - if (result.get() == Py_None) { - return bsl::optional(); - } - - // Extract tuple (mechanism, data) - if (!PyTuple_Check(result.get()) || PyTuple_Size(result.get()) != 2) { - BALL_LOG_ERROR - << "get_credential_data() must return (str, bytes) or None"; - return bsl::optional(); - } - - PyObject* mechanism_obj = PyTuple_GetItem(result.get(), 0); - PyObject* data_obj = PyTuple_GetItem(result.get(), 1); - - if (!PyUnicode_Check(mechanism_obj) || !PyBytes_Check(data_obj)) { - BALL_LOG_ERROR - << "get_credential_data() must return (str, bytes) or None"; - return bsl::optional(); - } - - // Convert Python str to C++ string - const char* mechanism_cstr = PyUnicode_AsUTF8(mechanism_obj); - bsl::string mechanism(mechanism_cstr); - - // Convert Python bytes to vector - char* data_ptr; - Py_ssize_t data_len; - PyBytes_AsStringAndSize(data_obj, &data_ptr, &data_len); - bsl::vector data(data_ptr, data_ptr + data_len); - - // Construct and move credential into optional - // (AuthnCredential is move-only) - bmqt::AuthnCredential credential(mechanism, data); - bsl::optional opt_credential( - bslmf::MovableRefUtil::move(credential)); - return opt_credential; - }; } { @@ -250,15 +266,22 @@ Session::Session( } Py_INCREF(d_error); Py_INCREF(d_broker_timeout_error); + Py_XINCREF(d_authn_credential_cb); } Session::~Session() { + BSLS_ASSERT(!d_started); + { + // Destroy the session first: it owns the copies of + // `AuthnCredentialCbFunctor`, which borrow `d_authn_credential_cb`. + pybmq::GilReleaseGuard gil_release_guard; + d_session_mp.reset(); + } + + Py_XDECREF(d_authn_credential_cb); Py_DECREF(d_broker_timeout_error); Py_DECREF(d_error); - BSLS_ASSERT(!d_started); - pybmq::GilReleaseGuard gil_release_guard; - d_session_mp.reset(); } PyObject* diff --git a/src/cpp/pybmq_session.h b/src/cpp/pybmq_session.h index 3a5ff74..c17ed3b 100644 --- a/src/cpp/pybmq_session.h +++ b/src/cpp/pybmq_session.h @@ -44,6 +44,7 @@ class Session bmqt::CompressionAlgorithmType::Enum d_message_compression_type; PyObject* d_error; PyObject* d_broker_timeout_error; + PyObject* d_authn_credential_cb; bslma::ManagedPtr d_session_mp; // NOT IMPLEMENTED From 43647689f93d890c9fdce19dea4ff450f2740a19 Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Tue, 1 Sep 2026 14:10:25 -0400 Subject: [PATCH 19/22] Move `authn_credential_provider` argument to avoid API break Signed-off-by: Patrick M. Niedzielski --- src/blazingmq/_session.py | 40 +++++++++++++++++++-------------------- 1 file changed, 20 insertions(+), 20 deletions(-) diff --git a/src/blazingmq/_session.py b/src/blazingmq/_session.py index ff259a2..9369aa0 100644 --- a/src/blazingmq/_session.py +++ b/src/blazingmq/_session.py @@ -308,11 +308,6 @@ class SessionOptions: healthy, `.HostUnhealthy` and `.HostHealthRestored` events with never be emitted, and the *suspends_on_bad_host_health* option of `QueueOptions` cannot be used. - authn_credential_provider (Optional[`~blazingmq.AuthnCredentialProvider`]): - An optional callable that returns authentication credentials as a - ``(mechanism, data)`` tuple of ``(str, bytes)``, or ``None`` if no - credentials are available. If not provided, no authentication - credentials are sent to the broker. num_processing_threads: The number of threads for the SDK to use for processing events. This defaults to 1. @@ -342,6 +337,11 @@ class SessionOptions: 96 bytes long. This is provided for libraries that are wrapping this SDK. Applications directly using the SDK are encouraged *NOT* to set this value. + authn_credential_provider (Optional[`~blazingmq.AuthnCredentialProvider`]): + An optional callable that returns authentication credentials as a + ``(mechanism, data)`` tuple of ``(str, bytes)``, or ``None`` if no + credentials are available. If not provided, no authentication + credentials are sent to the broker. """ def __init__( @@ -349,26 +349,26 @@ def __init__( message_compression_algorithm: Optional[CompressionAlgorithmType] = None, timeouts: Optional[Timeouts] = None, host_health_monitor: Union[BasicHealthMonitor, None] = (DefaultMonitor()), - authn_credential_provider: Optional[AuthnCredentialProvider] = ( - DefaultAuthnCredentialProvider() - ), num_processing_threads: Optional[int] = None, blob_buffer_size: Optional[int] = None, channel_high_watermark: Optional[int] = None, event_queue_watermarks: Optional[tuple[int, int]] = None, stats_dump_interval: Optional[float] = None, user_agent_prefix: Optional[bytes] = None, + authn_credential_provider: Optional[AuthnCredentialProvider] = ( + DefaultAuthnCredentialProvider() + ), ) -> None: self.message_compression_algorithm = message_compression_algorithm self.timeouts = timeouts self.host_health_monitor = host_health_monitor - self.authn_credential_provider = authn_credential_provider self.num_processing_threads = num_processing_threads self.blob_buffer_size = blob_buffer_size self.channel_high_watermark = channel_high_watermark self.event_queue_watermarks = event_queue_watermarks self.stats_dump_interval = stats_dump_interval self.user_agent_prefix = user_agent_prefix + self.authn_credential_provider = authn_credential_provider def __eq__(self, other: object) -> bool: if not isinstance(other, SessionOptions): @@ -377,13 +377,13 @@ def __eq__(self, other: object) -> bool: self.message_compression_algorithm == other.message_compression_algorithm and self.timeouts == other.timeouts and self.host_health_monitor == other.host_health_monitor - and self.authn_credential_provider == other.authn_credential_provider and self.num_processing_threads == other.num_processing_threads and self.blob_buffer_size == other.blob_buffer_size and self.channel_high_watermark == other.channel_high_watermark and self.event_queue_watermarks == other.event_queue_watermarks and self.stats_dump_interval == other.stats_dump_interval and self.user_agent_prefix == other.user_agent_prefix + and self.authn_credential_provider == other.authn_credential_provider ) def __ne__(self, other: object) -> bool: @@ -394,13 +394,13 @@ def __repr__(self) -> str: "message_compression_algorithm", "timeouts", "host_health_monitor", - "authn_credential_provider", "num_processing_threads", "blob_buffer_size", "channel_high_watermark", "event_queue_watermarks", "stats_dump_interval", "user_agent_prefix", + "authn_credential_provider", ) params = [] @@ -443,11 +443,6 @@ class Session: `.HostHealthRestored` events will never be emitted, and the *suspends_on_bad_host_health* option of `QueueOptions` cannot be used. - authn_credential_provider (Optional[`~blazingmq.AuthnCredentialProvider`]): - an optional callable that returns authentication credentials as a - ``(mechanism, data)`` tuple of ``(str, bytes)``, or ``None`` if no - credentials are available. If not provided, no authentication - credentials are sent to the broker. num_processing_threads: The number of threads for the SDK to use for processing events. This defaults to 1. blob_buffer_size: The size (in bytes) of the blob buffers to use. This @@ -473,6 +468,11 @@ class Session: must be at most 96 bytes long. This is provided for libraries that are wrapping this SDK. Applications directly using the SDK are encouraged *NOT* to set this value. + authn_credential_provider (Optional[`~blazingmq.AuthnCredentialProvider`]): + an optional callable that returns authentication credentials as a + ``(mechanism, data)`` tuple of ``(str, bytes)``, or ``None`` if no + credentials are available. If not provided, no authentication + credentials are sent to the broker. Raises: `~blazingmq.Error`: If the session start request was not successful. @@ -492,15 +492,15 @@ def __init__( ), timeout: Union[Timeouts, float] = DEFAULT_TIMEOUT, host_health_monitor: Union[BasicHealthMonitor, None] = (DefaultMonitor()), - authn_credential_provider: Optional[AuthnCredentialProvider] = ( - DefaultAuthnCredentialProvider() - ), num_processing_threads: Optional[int] = None, blob_buffer_size: Optional[int] = None, channel_high_watermark: Optional[int] = None, event_queue_watermarks: Optional[tuple[int, int]] = None, stats_dump_interval: Optional[float] = None, user_agent_prefix: Optional[bytes] = None, + authn_credential_provider: Optional[AuthnCredentialProvider] = ( + DefaultAuthnCredentialProvider() + ), ) -> None: if host_health_monitor is not None: if not isinstance(host_health_monitor, BasicHealthMonitor): @@ -602,13 +602,13 @@ def with_options( message_compression_algorithm=message_compression_algorithm, timeout=timeout, host_health_monitor=session_options.host_health_monitor, - authn_credential_provider=session_options.authn_credential_provider, num_processing_threads=session_options.num_processing_threads, blob_buffer_size=session_options.blob_buffer_size, channel_high_watermark=session_options.channel_high_watermark, event_queue_watermarks=session_options.event_queue_watermarks, stats_dump_interval=session_options.stats_dump_interval, user_agent_prefix=session_options.user_agent_prefix, + authn_credential_provider=session_options.authn_credential_provider, ) def open_queue( From 8c92a6ffa48877a1f112fe9693c2f6ee25397667 Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Tue, 1 Sep 2026 14:25:46 -0400 Subject: [PATCH 20/22] Fix: clarify docstring for `authn_credential_provider` Signed-off-by: Patrick M. Niedzielski --- src/blazingmq/_session.py | 20 ++++++++++++++------ src/blazingmq/_typing.py | 7 ++++--- 2 files changed, 18 insertions(+), 9 deletions(-) diff --git a/src/blazingmq/_session.py b/src/blazingmq/_session.py index 9369aa0..1c8ee87 100644 --- a/src/blazingmq/_session.py +++ b/src/blazingmq/_session.py @@ -339,9 +339,13 @@ class SessionOptions: to set this value. authn_credential_provider (Optional[`~blazingmq.AuthnCredentialProvider`]): An optional callable that returns authentication credentials as a - ``(mechanism, data)`` tuple of ``(str, bytes)``, or ``None`` if no - credentials are available. If not provided, no authentication - credentials are sent to the broker. + ``(mechanism, data)`` tuple of ``(str, bytes)``. It is called + each time the session authenticates with the broker, including on + reauthentication. If it returns ``None`` or raises, the + connection is closed: starting a session fails, while an + already-started session sees `.ConnectionLost` and then + reconnects, calling this callable again. If not provided, no + authentication credentials are sent to the broker. """ def __init__( @@ -470,9 +474,13 @@ class Session: are encouraged *NOT* to set this value. authn_credential_provider (Optional[`~blazingmq.AuthnCredentialProvider`]): an optional callable that returns authentication credentials as a - ``(mechanism, data)`` tuple of ``(str, bytes)``, or ``None`` if no - credentials are available. If not provided, no authentication - credentials are sent to the broker. + ``(mechanism, data)`` tuple of ``(str, bytes)``. It is called + each time the session authenticates with the broker, including on + reauthentication. If it returns ``None`` or raises, the + connection is closed: starting a session fails, while an + already-started session sees `.ConnectionLost` and then + reconnects, calling this callable again. If not provided, no + authentication credentials are sent to the broker. Raises: `~blazingmq.Error`: If the session start request was not successful. diff --git a/src/blazingmq/_typing.py b/src/blazingmq/_typing.py index 7c09d27..1110405 100644 --- a/src/blazingmq/_typing.py +++ b/src/blazingmq/_typing.py @@ -27,7 +27,8 @@ PropertyTypeDict = Mapping[str, PropertyType] AuthnCredentialProvider = Callable[[], Optional[tuple[str, bytes]]] -"""A callable that returns authentication credentials as a tuple of -``(mechanism, data)`` tuple of ``(str, bytes)``, or ``None`` if an -error occurs while obtaining credentials. +"""A callable that returns authentication credentials as a ``(mechanism, +data)`` tuple of ``(str, bytes)``, or ``None`` if an error occurs while +obtaining them, in which case authentication fails and the connection is +closed. """ From 9327cc7d23fc9f4f9cb2b500eb3d192bcb4f99d3 Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Tue, 1 Sep 2026 14:54:03 -0400 Subject: [PATCH 21/22] Style: Move UTF-8 encoding logic into Cython from C++ Signed-off-by: Patrick M. Niedzielski --- src/blazingmq/_ext.pyx | 24 ++++---- src/cpp/pybmq_session.cpp | 55 +++++++++---------- src/cpp/pybmq_session.h | 1 - src/declarations/bmq/bmqt.pxd | 12 +--- .../unit/test_authn_credential_cb_adapter.py | 44 ++++++++++++++- 5 files changed, 80 insertions(+), 56 deletions(-) diff --git a/src/blazingmq/_ext.pyx b/src/blazingmq/_ext.pyx index 1f97b09..6d66ae6 100644 --- a/src/blazingmq/_ext.pyx +++ b/src/blazingmq/_ext.pyx @@ -21,15 +21,12 @@ import weakref from bsl cimport optional from bsl cimport pair from bsl cimport shared_ptr -from bsl cimport string -from bsl cimport vector from bsl.bsls cimport TimeInterval from cpython.ceval cimport PyEval_InitThreads from libcpp cimport bool as cppbool from bmq.bmqa cimport ManualHostHealthMonitor from bmq.bmqt cimport AckResult -from bmq.bmqt cimport AuthnCredential from bmq.bmqt cimport CompressionAlgorithmType from bmq.bmqt cimport HostHealthState from bmq.bmqt cimport PropertyType @@ -158,31 +155,32 @@ cdef class FakeHostHealthMonitor: cdef class AuthnCredentialCbAdapter: - cdef object _callback # Store the Python callable + cdef object _callback def __cinit__(self, callback): self._callback = callback - # This method will be called by C++ code via PyObject_CallMethod - # Returns None for no credential, or (mechanism, data) tuple def get_credential_data(self): + """Call the provider and marshal its result for the C++ session. + + Called by ``pybmq::AuthnCredentialCbFunctor``. Returns the mechanism + and data as a tuple of ``bytes``, or `None` if credentials could not + be obtained, in which case authentication fails. + """ try: result = self._callback() if result is None: return None - if not isinstance(result, tuple) or len(result) != 2: - raise ValueError("callback must return (str, bytes) or None") - mechanism, data = result if not isinstance(mechanism, str) or not isinstance(data, bytes): - raise ValueError("callback must return (str, bytes) or None") + raise TypeError( + "authn_credential_provider must return (str, bytes) or None" + ) - # Return as-is, let C++ side handle conversion - return result + return mechanism.encode('utf-8'), data except Exception: - # Log error or handle as needed LOGGER.exception("Error in authentication credential callback") return None diff --git a/src/cpp/pybmq_session.cpp b/src/cpp/pybmq_session.cpp index ceca6ed..1748c68 100644 --- a/src/cpp/pybmq_session.cpp +++ b/src/cpp/pybmq_session.cpp @@ -22,17 +22,20 @@ #include #include -#include #include #include #include #include +#include +#include #include #include +#include #include #include #include +#include #include #include #include @@ -99,19 +102,16 @@ AuthnCredentialCbFunctor::AuthnCredentialCbFunctor(PyObject* callback) bsl::optional AuthnCredentialCbFunctor::operator()() const { - BALL_LOG_SET_CATEGORY("pybmq_session"); - pybmq::GilAcquireGuard guard; - // Call get_credential_data() method on the Python object + // The adapter validates the provider's result and reports any problem to + // the Python logger, so a failure here is just an empty credential. It + // hands back the mechanism and data already marshalled to `bytes`. bslma::ManagedPtr result = RefUtils::toManagedPtr( PyObject_CallMethod(d_callback_p, "get_credential_data", NULL)); if (!result) { - // Python exception occurred. Clear it before logging, so we - // don't re-enter Python via the BALL observer with it set. - PyErr_Print(); - BALL_LOG_ERROR << "Error calling get_credential_data()"; + PyErr_WriteUnraisable(d_callback_p); return bsl::optional(); } @@ -119,31 +119,26 @@ AuthnCredentialCbFunctor::operator()() const return bsl::optional(); } - // Extract tuple (mechanism, data) - if (!PyTuple_Check(result.get()) || PyTuple_Size(result.get()) != 2) { - BALL_LOG_ERROR << "get_credential_data() must return (str, bytes) or None"; - return bsl::optional(); - } - - PyObject* mechanism_obj = PyTuple_GetItem(result.get(), 0); - PyObject* data_obj = PyTuple_GetItem(result.get(), 1); - - if (!PyUnicode_Check(mechanism_obj) || !PyBytes_Check(data_obj)) { - BALL_LOG_ERROR << "get_credential_data() must return (str, bytes) or None"; + const char* mechanism_p; + Py_ssize_t mechanism_len; + const char* data_p; + Py_ssize_t data_len; + if (!PyArg_ParseTuple( + result.get(), + "y#y#", + &mechanism_p, + &mechanism_len, + &data_p, + &data_len)) + { + // The adapter broke its contract with us. + PyErr_WriteUnraisable(d_callback_p); return bsl::optional(); } - // Convert Python str to C++ string - const char* mechanism_cstr = PyUnicode_AsUTF8(mechanism_obj); - bsl::string mechanism(mechanism_cstr); - - // Convert Python bytes to vector - char* data_ptr; - Py_ssize_t data_len; - PyBytes_AsStringAndSize(data_obj, &data_ptr, &data_len); - bsl::vector data(data_ptr, data_ptr + data_len); - - bmqt::AuthnCredential credential(mechanism, data); + bmqt::AuthnCredential credential( + bsl::string_view(mechanism_p, mechanism_len), + bsl::vector(data_p, data_p + data_len)); return bsl::optional( bslmf::MovableRefUtil::move(credential)); } diff --git a/src/cpp/pybmq_session.h b/src/cpp/pybmq_session.h index c17ed3b..8d3823e 100644 --- a/src/cpp/pybmq_session.h +++ b/src/cpp/pybmq_session.h @@ -23,7 +23,6 @@ #include #include -#include #include #include diff --git a/src/declarations/bmq/bmqt.pxd b/src/declarations/bmq/bmqt.pxd index 5f05d12..07e27e9 100644 --- a/src/declarations/bmq/bmqt.pxd +++ b/src/declarations/bmq/bmqt.pxd @@ -1,4 +1,4 @@ -# Copyright 2019-2026 Bloomberg Finance L.P. +# Copyright 2019-2023 Bloomberg Finance L.P. # SPDX-License-Identifier: Apache-2.0 # # Licensed under the Apache License, Version 2.0 (the "License"); @@ -13,8 +13,6 @@ # See the License for the specific language governing permissions and # limitations under the License. -from bsl cimport string -from bsl cimport vector from libcpp cimport bool @@ -75,11 +73,3 @@ cdef extern from "bmqt_queueoptions.h" namespace "BloombergLP::bmqt::QueueOption int k_DEFAULT_MAX_UNCONFIRMED_BYTES int k_DEFAULT_CONSUMER_PRIORITY bool k_DEFAULT_SUSPENDS_ON_BAD_HOST_HEALTH - -cdef extern from "bmqt_authncredential.h" namespace "BloombergLP::bmqt" nogil: - cdef cppclass AuthnCredential: - AuthnCredential() except + - AuthnCredential(const AuthnCredential&) except + - AuthnCredential(const string& mechanism, const vector[char]& data) except + - const string& mechanism() const - const vector[char]& data() const diff --git a/tests/unit/test_authn_credential_cb_adapter.py b/tests/unit/test_authn_credential_cb_adapter.py index 4e639e2..54b2d11 100644 --- a/tests/unit/test_authn_credential_cb_adapter.py +++ b/tests/unit/test_authn_credential_cb_adapter.py @@ -27,7 +27,49 @@ def provider(): result = adapter.get_credential_data() # THEN - assert result == ("mechanism", b"data") + assert result == (b"mechanism", b"data") + + +def test_mechanism_is_encoded_as_utf8(): + # GIVEN + def provider(): + return ("mécanisme", b"data") + + adapter = AuthnCredentialCbAdapter(provider) + + # WHEN + result = adapter.get_credential_data() + + # THEN + assert result == ("mécanisme".encode("utf-8"), b"data") + + +def test_mechanism_not_encodable(): + # GIVEN a mechanism holding a lone surrogate, which has no UTF-8 encoding + def provider(): + return ("\ud800", b"data") + + adapter = AuthnCredentialCbAdapter(provider) + + # WHEN + result = adapter.get_credential_data() + + # THEN + assert result is None + + +def test_mechanism_with_embedded_nul_is_not_truncated(): + # GIVEN + def provider(): + return ("PLAIN\x00extra", b"data") + + adapter = AuthnCredentialCbAdapter(provider) + + # WHEN + result = adapter.get_credential_data() + + # THEN + assert result == (b"PLAIN\x00extra", b"data") def test_none_return(): From 9573080a0983a074a5b561baa5b52bd890fa4e9b Mon Sep 17 00:00:00 2001 From: "Patrick M. Niedzielski" Date: Tue, 1 Sep 2026 15:14:13 -0400 Subject: [PATCH 22/22] Fix: Simplify authn callback guard and bump copyright years Remove the redundant `has_auth_callback` flag in favour of testing the `bsl::function` directly, and bump copyright years on every file the branch modifies. --- src/blazingmq/__init__.py | 2 +- src/blazingmq/_ext.pyi | 2 +- src/blazingmq/_ext.pyx | 2 +- src/blazingmq/_session.py | 2 +- src/blazingmq/_typing.py | 2 +- src/cpp/pybmq_session.cpp | 6 ++---- src/cpp/pybmq_session.h | 2 +- src/declarations/pybmq.pxd | 2 +- tests/unit/test_session.py | 2 +- tests/unit/test_session_options.py | 2 +- 10 files changed, 11 insertions(+), 13 deletions(-) diff --git a/src/blazingmq/__init__.py b/src/blazingmq/__init__.py index 28d81ef..e630190 100644 --- a/src/blazingmq/__init__.py +++ b/src/blazingmq/__init__.py @@ -1,4 +1,4 @@ -# Copyright 2019-2023 Bloomberg Finance L.P. +# Copyright 2019-2026 Bloomberg Finance L.P. # SPDX-License-Identifier: Apache-2.0 # # Licensed under the Apache License, Version 2.0 (the "License"); diff --git a/src/blazingmq/_ext.pyi b/src/blazingmq/_ext.pyi index 7b94abc..a035f66 100644 --- a/src/blazingmq/_ext.pyi +++ b/src/blazingmq/_ext.pyi @@ -1,4 +1,4 @@ -# Copyright 2019-2023 Bloomberg Finance L.P. +# Copyright 2019-2026 Bloomberg Finance L.P. # SPDX-License-Identifier: Apache-2.0 # # Licensed under the Apache License, Version 2.0 (the "License"); diff --git a/src/blazingmq/_ext.pyx b/src/blazingmq/_ext.pyx index 6d66ae6..e0b2be7 100644 --- a/src/blazingmq/_ext.pyx +++ b/src/blazingmq/_ext.pyx @@ -1,4 +1,4 @@ -# Copyright 2019-2023 Bloomberg Finance L.P. +# Copyright 2019-2026 Bloomberg Finance L.P. # SPDX-License-Identifier: Apache-2.0 # # Licensed under the Apache License, Version 2.0 (the "License"); diff --git a/src/blazingmq/_session.py b/src/blazingmq/_session.py index 1c8ee87..16afbb0 100644 --- a/src/blazingmq/_session.py +++ b/src/blazingmq/_session.py @@ -1,4 +1,4 @@ -# Copyright 2019-2023 Bloomberg Finance L.P. +# Copyright 2019-2026 Bloomberg Finance L.P. # SPDX-License-Identifier: Apache-2.0 # # Licensed under the Apache License, Version 2.0 (the "License"); diff --git a/src/blazingmq/_typing.py b/src/blazingmq/_typing.py index 1110405..c6fe00a 100644 --- a/src/blazingmq/_typing.py +++ b/src/blazingmq/_typing.py @@ -1,4 +1,4 @@ -# Copyright 2019-2023 Bloomberg Finance L.P. +# Copyright 2019-2026 Bloomberg Finance L.P. # SPDX-License-Identifier: Apache-2.0 # # Licensed under the Apache License, Version 2.0 (the "License"); diff --git a/src/cpp/pybmq_session.cpp b/src/cpp/pybmq_session.cpp index 1748c68..5540e89 100644 --- a/src/cpp/pybmq_session.cpp +++ b/src/cpp/pybmq_session.cpp @@ -1,4 +1,4 @@ -// Copyright 2019-2023 Bloomberg Finance L.P. +// Copyright 2019-2026 Bloomberg Finance L.P. // SPDX-License-Identifier: Apache-2.0 // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -183,12 +183,10 @@ Session::Session( d_message_compression_type = config.message_compression_type; bmqt::SessionOptions::AuthnCredentialCb cpp_callback; - bool has_auth_callback = false; if (authn_credential_cb != NULL && authn_credential_cb != Py_None) { d_authn_credential_cb = authn_credential_cb; cpp_callback = AuthnCredentialCbFunctor(d_authn_credential_cb); - has_auth_callback = true; } { @@ -216,7 +214,7 @@ Session::Session( config.event_queue_watermarks.value().second); } - if (has_auth_callback) { + if (cpp_callback) { options.setAuthnCredentialCb(cpp_callback); } diff --git a/src/cpp/pybmq_session.h b/src/cpp/pybmq_session.h index 8d3823e..01e08ea 100644 --- a/src/cpp/pybmq_session.h +++ b/src/cpp/pybmq_session.h @@ -1,4 +1,4 @@ -// Copyright 2019-2023 Bloomberg Finance L.P. +// Copyright 2019-2026 Bloomberg Finance L.P. // SPDX-License-Identifier: Apache-2.0 // // Licensed under the Apache License, Version 2.0 (the "License"); diff --git a/src/declarations/pybmq.pxd b/src/declarations/pybmq.pxd index d51a45d..67bc0cb 100644 --- a/src/declarations/pybmq.pxd +++ b/src/declarations/pybmq.pxd @@ -1,4 +1,4 @@ -# Copyright 2019-2023 Bloomberg Finance L.P. +# Copyright 2019-2026 Bloomberg Finance L.P. # SPDX-License-Identifier: Apache-2.0 # # Licensed under the Apache License, Version 2.0 (the "License"); diff --git a/tests/unit/test_session.py b/tests/unit/test_session.py index 9b2258f..349153b 100644 --- a/tests/unit/test_session.py +++ b/tests/unit/test_session.py @@ -1,4 +1,4 @@ -# Copyright 2019-2023 Bloomberg Finance L.P. +# Copyright 2019-2026 Bloomberg Finance L.P. # SPDX-License-Identifier: Apache-2.0 # # Licensed under the Apache License, Version 2.0 (the "License"); diff --git a/tests/unit/test_session_options.py b/tests/unit/test_session_options.py index 43f4026..1c31cee 100644 --- a/tests/unit/test_session_options.py +++ b/tests/unit/test_session_options.py @@ -1,4 +1,4 @@ -# Copyright 2019-2023 Bloomberg Finance L.P. +# Copyright 2019-2026 Bloomberg Finance L.P. # SPDX-License-Identifier: Apache-2.0 # # Licensed under the Apache License, Version 2.0 (the "License");