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

news/120.feature.rst

@@ -0,0 +1 @@
+Added the ``authn_credential_provider`` option to ``Session`` and ``SessionOptions``, a callable that supplies authentication credentials as a ``(mechanism, data)`` tuple each time the session authenticates with the broker.

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,
         user_agent_prefix: bytes = b"",
     ) -> None: ...
     def stop(self) -> None: ...

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");
@@ -157,6 +157,42 @@ 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
+
+            try:
+                mechanism, data = result
+            except (TypeError, ValueError):
+                pass            # handled below
+            else:
+                if isinstance(mechanism, str) and isinstance(data, bytes):
+                    return mechanism.encode('utf-8'), data
+
+            # If we got an exception above, or the types were wrong:
+            raise TypeError(
+                "authn_credential_provider must return (str, bytes) or None"
+            )
+
+        except Exception:
+            LOGGER.exception("Error in authentication credential callback")
+            return None
+
+
 cdef class Session:
     cdef object __weakref__
     cdef shared_ptr[NativeSession] _session
@@ -178,6 +214,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,
         user_agent_prefix: bytes = b"",
     ) -> None:
@@ -252,6 +289,7 @@ cdef class Session:
             session_cb,
             message_cb,
             ack_cb,
+            authn_credential_cb,
             config,
             fake_host_health_monitor_sp,
             Error,

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");
@@ -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
@@ -38,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
@@ -54,6 +56,10 @@ def DefaultMonitor() -> Union[BasicHealthMonitor, None]:
     return None
 
 
+def DefaultAuthnCredentialProvider() -> Optional[AuthnCredentialProvider]:
+    return None
+
+
 DEFAULT_TIMEOUT = DefaultTimeoutType()
 KNOWN_MONITORS = ("blazingmq.BasicHealthMonitor",)
 
@@ -349,6 +355,15 @@ 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)``.  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__(
@@ -362,6 +377,9 @@ def __init__(
         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
@@ -372,6 +390,7 @@ def __init__(
         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):
@@ -386,6 +405,7 @@ def __eq__(self, other: object) -> bool:
             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:
@@ -402,6 +422,7 @@ def __repr__(self) -> str:
             "event_queue_watermarks",
             "stats_dump_interval",
             "user_agent_prefix",
+            "authn_credential_provider",
         )
 
         params = []
@@ -470,6 +491,15 @@ 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)``.  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.
@@ -496,6 +526,9 @@ def __init__(
         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):
@@ -506,6 +539,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
 
@@ -537,6 +575,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,
             user_agent_prefix=_make_user_agent_prefix(user_agent_prefix),
         )
         self._ext.set_owned_by_session()
@@ -600,6 +639,7 @@ def with_options(
             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(

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.
+"""