Add authentication credential callback support

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

docs/api_reference.rst

@@ -187,6 +187,8 @@ Testing Utilities
 Helper Types
 ============
 
+.. autoclass:: blazingmq.AuthnCredentialProvider
+
 .. autoclass:: blazingmq.PropertyTypeDict
 
 .. autoclass:: blazingmq.PropertyValueDict

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");
@@ -27,13 +27,15 @@
 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
 
 __all__ = [
     "Ack",
     "AckStatus",
+    "AuthnCredentialProvider",
     "BasicHealthMonitor",
     "CompressionAlgorithmType",
     "Error",

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");
@@ -37,6 +37,9 @@ class FakeHostHealthMonitor:
     def set_healthy(self) -> None: ...
     def set_unhealthy(self) -> None: ...
 
+class AuthnCredentialCbAdapter:
+    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,
+        authn_credential_cb: Optional[AuthnCredentialCbAdapter] = None,
     ) -> None: ...
     def stop(self) -> None: ...
     def open_queue_sync(

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");
@@ -37,6 +37,7 @@ from bmq.bmqt cimport k_DEFAULT_MAX_UNCONFIRMED_MESSAGES
 from bmq.bmqt cimport k_DEFAULT_SUSPENDS_ON_BAD_HOST_HEALTH
 from pybmq cimport BallUtil
 from pybmq cimport Session as NativeSession
+from pybmq cimport SessionConfig
 
 from typing import Optional
 
@@ -153,6 +154,37 @@ cdef class FakeHostHealthMonitor:
             self._monitor.get().setState(HostHealthState.e_UNHEALTHY)
 
 
+cdef class AuthnCredentialCbAdapter:
+    cdef object _callback
+
+    def __cinit__(self, callback):
+        self._callback = callback
+
+    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
+
+            mechanism, data = result
+            if not isinstance(mechanism, str) or not isinstance(data, bytes):
+                raise TypeError(
+                    "authn_credential_provider must return (str, bytes) or None"
+                )
+
+            return mechanism.encode('utf-8'), data
+
+        except Exception:
+            LOGGER.exception("Error in authentication credential callback")
+            return None
+
+
 cdef class Session:
     cdef object __weakref__
     cdef NativeSession* _session
@@ -174,6 +206,7 @@ cdef class Session:
         timeouts: _timeouts.Timeouts = _timeouts.Timeouts(),
         monitor_host_health: bool = False,
         fake_host_health_monitor: FakeHostHealthMonitor = None,
+        authn_credential_cb: AuthnCredentialCbAdapter = None,
         _mock: Optional[object] = None,
     ) -> None:
         cdef shared_ptr[ManualHostHealthMonitor] fake_host_health_monitor_sp
@@ -221,24 +254,30 @@ cdef class Session:
         cdef char *c_broker_uri = broker
         script_name = _script_name.get_script_name()
         cdef char *c_script_name = script_name
+
+        cdef SessionConfig config
+        config.broker_uri = c_broker_uri
+        config.script_name = c_script_name
+        config.message_compression_type = (
+            COMPRESSION_ALGO_FROM_PY_MAPPING[message_compression_algorithm])
+        config.num_processing_threads = c_num_processing_threads
+        config.blob_buffer_size = c_blob_buffer_size
+        config.channel_high_watermark = c_channel_high_watermark
+        config.event_queue_watermarks = c_event_queue_watermarks
+        config.stats_dump_interval = c_stats_dump_interval
+        config.connect_timeout = c_connect_timeout
+        config.disconnect_timeout = c_disconnect_timeout
+        config.open_queue_timeout = c_open_queue_timeout
+        config.configure_queue_timeout = c_configure_queue_timeout
+        config.close_queue_timeout = c_close_queue_timeout
+        config.monitor_host_health = monitor_host_health
+
         self._session = new NativeSession(
             session_cb,
             message_cb,
             ack_cb,
-            c_broker_uri,
-            c_script_name,
-            COMPRESSION_ALGO_FROM_PY_MAPPING[message_compression_algorithm],
-            c_num_processing_threads,
-            c_blob_buffer_size,
-            c_channel_high_watermark,
-            c_event_queue_watermarks,
-            c_stats_dump_interval,
-            c_connect_timeout,
-            c_disconnect_timeout,
-            c_open_queue_timeout,
-            c_configure_queue_timeout,
-            c_close_queue_timeout,
-            monitor_host_health,
+            authn_credential_cb,
+            config,
             fake_host_health_monitor_sp,
             Error,
             BrokerTimeoutError,

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");
@@ -25,6 +25,7 @@
 from . import _six as six
 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
@@ -36,6 +37,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
@@ -52,6 +54,10 @@ def DefaultMonitor() -> Union[BasicHealthMonitor, None]:
     return None
 
 
+def DefaultAuthnCredentialProvider() -> Optional[AuthnCredentialProvider]:
+    return None
+
+
 DEFAULT_TIMEOUT = DefaultTimeoutType()
 KNOWN_MONITORS = ("blazingmq.BasicHealthMonitor",)
 
@@ -288,6 +294,15 @@ class SessionOptions:
             0, disable the recurring dump of stats (final stats are always
             dumped at the end of the session).  The default is 5min; the value
             must be a multiple of 30s, in the range ``[0s - 60min]``.
+        authn_credential_provider (Optional[`~blazingmq.AuthnCredentialProvider`]):
+            An optional callable that returns authentication credentials as a
+            ``(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__(
@@ -300,6 +315,9 @@ def __init__(
         channel_high_watermark: Optional[int] = None,
         event_queue_watermarks: Optional[tuple[int, int]] = None,
         stats_dump_interval: Optional[float] = None,
+        authn_credential_provider: Optional[AuthnCredentialProvider] = (
+            DefaultAuthnCredentialProvider()
+        ),
     ) -> None:
         self.message_compression_algorithm = message_compression_algorithm
         self.timeouts = timeouts
@@ -309,6 +327,7 @@ def __init__(
         self.channel_high_watermark = channel_high_watermark
         self.event_queue_watermarks = event_queue_watermarks
         self.stats_dump_interval = stats_dump_interval
+        self.authn_credential_provider = authn_credential_provider
 
     def __eq__(self, other: object) -> bool:
         if not isinstance(other, SessionOptions):
@@ -322,6 +341,7 @@ def __eq__(self, other: object) -> bool:
             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.authn_credential_provider == other.authn_credential_provider
         )
 
     def __ne__(self, other: object) -> bool:
@@ -337,6 +357,7 @@ def __repr__(self) -> str:
             "channel_high_watermark",
             "event_queue_watermarks",
             "stats_dump_interval",
+            "authn_credential_provider",
         )
 
         params = []
@@ -399,6 +420,15 @@ class Session:
             stats are always dumped at the end of the session).  The default is
             5min; the value must be a multiple of 30s, in the range
             ``[0s - 60min]``.
+        authn_credential_provider (Optional[`~blazingmq.AuthnCredentialProvider`]):
+            an optional callable that returns authentication credentials as a
+            ``(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.
@@ -423,6 +453,9 @@ def __init__(
         channel_high_watermark: Optional[int] = None,
         event_queue_watermarks: Optional[tuple[int, int]] = None,
         stats_dump_interval: Optional[float] = None,
+        authn_credential_provider: Optional[AuthnCredentialProvider] = (
+            DefaultAuthnCredentialProvider()
+        ),
     ) -> None:
         if host_health_monitor is not None:
             if not isinstance(host_health_monitor, BasicHealthMonitor):
@@ -433,6 +466,11 @@ def __init__(
 
         monitor_host_health = host_health_monitor is not None
         fake_host_health_monitor = getattr(host_health_monitor, "_monitor", None)
+        authn_credential_cb = (
+            AuthnCredentialCbAdapter(authn_credential_provider)
+            if authn_credential_provider is not None
+            else None
+        )
 
         self._has_no_on_message = on_message is None
 
@@ -459,6 +497,7 @@ def __init__(
             timeouts=_validate_timeouts(timeout),
             monitor_host_health=monitor_host_health,
             fake_host_health_monitor=fake_host_health_monitor,
+            authn_credential_cb=authn_credential_cb,
         )
         self._ext.set_owned_by_session()
 
@@ -503,34 +542,24 @@ def with_options(
         if message_compression_algorithm is None:
             message_compression_algorithm = CompressionAlgorithmType.NONE
 
-        if session_options.timeouts is None:
-            return cls(
-                on_session_event,
-                on_message,
-                broker,
-                message_compression_algorithm,
-                DEFAULT_TIMEOUT,
-                session_options.host_health_monitor,
-                session_options.num_processing_threads,
-                session_options.blob_buffer_size,
-                session_options.channel_high_watermark,
-                session_options.event_queue_watermarks,
-                session_options.stats_dump_interval,
-            )
-        else:
-            return cls(
-                on_session_event,
-                on_message,
-                broker,
-                message_compression_algorithm,
-                session_options.timeouts,
-                session_options.host_health_monitor,
-                session_options.num_processing_threads,
-                session_options.blob_buffer_size,
-                session_options.channel_high_watermark,
-                session_options.event_queue_watermarks,
-                session_options.stats_dump_interval,
-            )
+        timeout: Union[Timeouts, float] = DEFAULT_TIMEOUT
+        if session_options.timeouts is not None:
+            timeout = session_options.timeouts
+
+        return cls(
+            on_session_event,
+            on_message=on_message,
+            broker=broker,
+            message_compression_algorithm=message_compression_algorithm,
+            timeout=timeout,
+            host_health_monitor=session_options.host_health_monitor,
+            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,
+            authn_credential_provider=session_options.authn_credential_provider,
+        )
 
     def open_queue(
         self,

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");
@@ -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,10 @@
 PropertyValueDict = Mapping[str, PropertyValueType]
 
 PropertyTypeDict = Mapping[str, PropertyType]
+
+AuthnCredentialProvider = Callable[[], Optional[tuple[str, bytes]]]
+"""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.
+"""