WIP: bmqbrkrcfg.json and authn timeout

Author: emelialei88

src/applications/bmqbrkr/etc/bmqbrkrcfg.json

@@ -92,6 +92,23 @@
         },
         "bmqconfConfig": {
             "cacheTTLSeconds": 30
+        },
+        "plugins": {
+            "libraries": ["/Users/wlei29/Workspace/blazingmq/build/blazingmq/src/plugins/"],
+            "enabled": ["PassAuthenticator"]
+        },
+        "authentication": {
+            "plugins": [
+                {
+                    "name": "FailAuthenticator",
+                    "configs": []
+                },
+                {
+                    "name": "PassAuthenticator",
+                    "configs": []
+                }
+            ],
+            "fallbackPrincipal": "defaultFallbackPrincipal"
         }
     }
 }

src/groups/mqb/mqba/mqba_authenticator.cpp

@@ -14,6 +14,7 @@
 // limitations under the License.
 
 // mqba_authenticator.h                                           -*-C++-*-
+#include <bdlmt_timereventscheduler.h>
 #include <mqba_authenticator.h>
 
 #include <mqbscm_version.h>
@@ -55,9 +56,6 @@ namespace mqba {
 
 namespace {
 BALL_LOG_SET_NAMESPACE_CATEGORY("MQBA.AUTHENTICATOR");
-
-const int k_AUTHENTICATION_READTIMEOUT = 3 * 60;  // 3 minutes
-
 }
 
 // -------------------
@@ -224,7 +222,6 @@ void Authenticator::authenticate(
                                                &result,
                                                authenticateRequest.mechanism(),
                                                authenticationData);
-
     if (rc != 0) {
         BALL_LOG_ERROR << "Authentication failed for connection '"
                        << channel->peerUri() << "' with mechanism '"
@@ -282,7 +279,10 @@ void Authenticator::authenticate(
                 rc,
                 sendResponseErrorStream.str(),
                 bsl::shared_ptr<mqbnet::Session>());
+            return;  // RETURN
         }
+
+        setTimer(context, result, channel);
     }
 }
 
@@ -338,7 +338,7 @@ void Authenticator::reAuthenticate(
                                                authenticateRequest.mechanism(),
                                                authenticationData);
     if (rc != 0) {
-        BALL_LOG_ERROR << "Authentication failed for connection '"
+        BALL_LOG_ERROR << "Reauthentication failed for connection '"
                        << channel->peerUri() << "' with mechanism '"
                        << authenticateRequest.mechanism() << "' [rc: " << rc
                        << ", error: " << authenticationErrorStream.str()
@@ -356,13 +356,14 @@ void Authenticator::reAuthenticate(
                                   channel,
                                   context->authenticationEncodingType());
 
-        bmqio::Status status(bmqio::StatusCategory::e_GENERIC_ERROR,
-                             "reAuthenticationError",
-                             rc,
-                             d_allocator_p);
-        channel->close(status);
+        cleanupOnError(rc, "reauthenticationError", context, channel);
     }
     else {
+        // Authentication succeeded
+        BALL_LOG_INFO << "Reauthentication succeeded for connection '"
+                      << channel->peerUri() << "' with mechanism '"
+                      << authenticateRequest.mechanism() << "'";
+
         response.status().code()     = 0;
         response.status().category() = bmqp_ctrlmsg::StatusCategory::E_SUCCESS;
         response.lifetimeMs()        = result->lifetimeMs();
@@ -375,11 +376,7 @@ void Authenticator::reAuthenticate(
             BALL_LOG_ERROR << "Failed to set (re)authentication state for '"
                            << channel->peerUri()
                            << "' to 'e_AUTHENTICATED' from 'e_AUTHENTICATING'";
-            bmqio::Status status(bmqio::StatusCategory::e_GENERIC_ERROR,
-                                 "reAuthenticationError",
-                                 rc,
-                                 d_allocator_p);
-            channel->close(status);
+            cleanupOnError(rc, "reauthenticationError", context, channel);
             return;  // RETURN
         }
 
@@ -391,12 +388,11 @@ void Authenticator::reAuthenticate(
                                        context->authenticationEncodingType());
 
         if (rc != 0) {
-            bmqio::Status status(bmqio::StatusCategory::e_GENERIC_ERROR,
-                                 "reAuthenticationError",
-                                 rc,
-                                 d_allocator_p);
-            channel->close(status);
+            cleanupOnError(rc, "reauthenticationError", context, channel);
+            return;  // RETURN
         }
+
+        setTimer(context, result, channel);
     }
 }
 
@@ -411,6 +407,8 @@ Authenticator::Authenticator(
                100,                                          // max threads
                bsls::TimeInterval(120).totalMilliseconds(),  // idle time
                allocator)
+, d_scheduler(bdlmt::TimerEventScheduler(bsls::SystemClockType::e_MONOTONIC,
+                                         allocator))
 , d_blobSpPool_p(blobSpPool)
 , d_allocator_p(allocator)
 {
@@ -432,6 +430,15 @@ int Authenticator::start(bsl::ostream& errorDescription)
         return rc;  // RETURN
     }
 
+    rc = d_scheduler.start();
+    if (rc != 0) {
+        errorDescription << "Failed to start TimerEventScheduler for "
+                            "Authenticator [rc: "
+                         << rc << "]";
+        d_threadPool.stop();
+        return rc;  // RETURN
+    }
+
     return 0;
 }
 
@@ -514,6 +521,82 @@ int Authenticator::handleReauthentication(
     return rc;
 }
 
+void Authenticator::timeout(const ChannelSp& channel)
+{
+    BALL_LOG_DEBUG << "Authentication timeout for channel: "
+                   << channel->peerUri();
+
+    bmqio::Status status(bmqio::StatusCategory::e_TIMEOUT,
+                         "authenticationTimeout",
+                         -1,
+                         d_allocator_p);
+    channel->close(status);
+}
+
+void Authenticator::setTimer(const AuthenticationContextSp& context,
+                             const AuthenticationResultSp&  result,
+                             const ChannelSp&               channel)
+{
+    if (!result->lifetimeMs().has_value()) {
+        BALL_LOG_INFO
+            << "Lifetime not set for authentication timer for channel: "
+            << channel->peerUri() << ", this indicates an infinite lifetime.";
+        return;  // RETURN
+    }
+
+    bslmt::LockGuard<bslmt::Mutex> guard(
+        &context->authenticationTimerHandleMutex());  // MUTEX LOCKED
+
+    bdlmt::TimerEventScheduler::Handle handle =
+        context->authenticationTimerHandle();
+
+    // If the timer handle is invalid, it means that this is the initial
+    // authentication and the timer has not been set yet.
+    if (handle == bdlmt::TimerEventScheduler::INVALID_HANDLE) {
+        handle = d_scheduler.scheduleEvent(
+            bsls::TimeInterval(result->lifetimeMs().value()),
+            bdlf::BindUtil::bind(&Authenticator::timeout, this, channel));
+        context->setAuthenticationTimerHandle(handle);
+    }
+    else {
+        d_scheduler.cancelEvent(handle);
+
+        int rc = d_scheduler.rescheduleEvent(
+            handle,
+            bsls::TimeInterval(result->lifetimeMs().value()));
+
+        if (rc != 0) {
+            BALL_LOG_ERROR << "Failed to reschedule authentication timer for '"
+                           << channel->peerUri() << "' [rc: " << rc
+                           << ", lifetime: " << result->lifetimeMs().value()
+                           << "]";
+            bmqio::Status status(bmqio::StatusCategory::e_GENERIC_ERROR,
+                                 "reAuthenticationError",
+                                 rc,
+                                 d_allocator_p);
+            channel->close(status);
+        }
+    }
+}
+
+void Authenticator::cleanupOnError(int                errorCode,
+                                   const bsl::string& errorDescription,
+                                   const AuthenticationContextSp& context,
+                                   const ChannelSp&               channel)
+{
+    {
+        bslmt::LockGuard<bslmt::Mutex> guard(
+            &context->authenticationTimerHandleMutex());  // MUTEX LOCKED
+        d_scheduler.cancelEvent(context->authenticationTimerHandle());
+    }
+
+    bmqio::Status status(bmqio::StatusCategory::e_GENERIC_ERROR,
+                         errorDescription,
+                         errorCode,
+                         d_allocator_p);
+    channel->close(status);
+}
+
 int Authenticator::authenticationOutboundOrReverse(
     const AuthenticationContextSp& context)
 {

src/groups/mqb/mqba/mqba_authenticator.h

@@ -32,6 +32,7 @@
 /// are called only from there.  It is not thread safe.
 
 // MQB
+#include "mqbplug_authenticator.h"
 #include <mqbauthn_authenticationcontroller.h>
 #include <mqbconfm_messages.h>
 #include <mqbnet_authenticationcontext.h>
@@ -47,6 +48,7 @@
 #include <bdlbb_blob.h>
 #include <bdlcc_sharedobjectpool.h>
 #include <bdlmt_threadpool.h>
+#include <bdlmt_timereventscheduler.h>
 #include <bsl_memory.h>
 #include <bsl_ostream.h>
 #include <bslma_allocator.h>
@@ -90,6 +92,11 @@ class Authenticator : public mqbnet::Authenticator {
     typedef bsl::shared_ptr<mqbnet::InitialConnectionContext>
         InitialConnectionContextSp;
 
+    typedef bsl::shared_ptr<mqbplug::AuthenticationResult>
+        AuthenticationResultSp;
+
+    typedef bsl::shared_ptr<bmqio::Channel> ChannelSp;
+
   private:
     // DATA
 
@@ -98,6 +105,8 @@ class Authenticator : public mqbnet::Authenticator {
 
     bdlmt::ThreadPool d_threadPool;
 
+    bdlmt::TimerEventScheduler d_scheduler;
+
     BlobSpPool* d_blobSpPool_p;
 
     /// Allocator to use.
@@ -193,6 +202,17 @@ class Authenticator : public mqbnet::Authenticator {
                                const AuthenticationContextSp& context,
                                const bsl::shared_ptr<bmqio::Channel>& channel);
 
+    void timeout(const bsl::shared_ptr<bmqio::Channel>& channel);
+
+    void setTimer(const AuthenticationContextSp& context,
+                  const AuthenticationResultSp&  result,
+                  const ChannelSp&               channel);
+
+    void cleanupOnError(int                            errorCode,
+                        const bsl::string&             errorDescription,
+                        const AuthenticationContextSp& context,
+                        const ChannelSp&               channel);
+
   public:
     // TRAITS
     BSLMF_NESTED_TRAIT_DECLARATION(Authenticator, bslma::UsesBslmaAllocator)

src/groups/mqb/mqbnet/mqbnet_authenticationcontext.cpp

@@ -39,7 +39,11 @@ AuthenticationContext::AuthenticationContext(
     bool                                       isReversed,
     State                                      state,
     ConnectionType::Enum                       connectionType)
-: d_initialConnectionContext_p(initialConnectionContext)
+: d_authenticationResultSp()
+, d_authenticationTimerHandle(bdlmt::TimerEventScheduler::INVALID_HANDLE)
+, d_timerHandleMutex()
+, d_mutex()
+, d_initialConnectionContext_p(initialConnectionContext)
 , d_authenticationMessage(authenticationMessage)
 , d_authenticationEncodingType(authenticationEncodingType)
 , d_reAuthenticateCb(reAuthenticateCb)
@@ -59,6 +63,13 @@ AuthenticationContext& AuthenticationContext::setAuthenticationResult(
     return *this;
 }
 
+AuthenticationContext& AuthenticationContext::setAuthenticationTimerHandle(
+    bdlmt::TimerEventScheduler::Handle value)
+{
+    d_authenticationTimerHandle = value;
+    return *this;
+}
+
 AuthenticationContext& AuthenticationContext::setInitialConnectionContext(
     InitialConnectionContext* value)
 {
@@ -113,6 +124,17 @@ AuthenticationContext::authenticationResult() const
     return d_authenticationResultSp;
 }
 
+bdlmt::TimerEventScheduler::Handle
+AuthenticationContext::authenticationTimerHandle() const
+{
+    return d_authenticationTimerHandle;
+}
+
+bslmt::Mutex& AuthenticationContext::authenticationTimerHandleMutex()
+{
+    return d_timerHandleMutex;
+}
+
 InitialConnectionContext*
 AuthenticationContext::initialConnectionContext() const
 {

src/groups/mqb/mqbnet/mqbnet_authenticationcontext.h

@@ -23,13 +23,14 @@
 ///
 
 // MQB
-#include "bmqp_protocol.h"
 #include <mqbnet_initialconnectioncontext.h>
 
 // BMQ
 #include <bmqp_ctrlmsg_messages.h>
+#include <bmqp_protocol.h>
 
 // BDE
+#include <bdlmt_timereventscheduler.h>
 #include <bslmt_mutex.h>
 #include <bsls_atomic.h>
 
@@ -63,6 +64,12 @@ class AuthenticationContext {
     /// during re-authentication.
     bsl::shared_ptr<mqbplug::AuthenticationResult> d_authenticationResultSp;
 
+    /// The timer handle for the authentication timeout.
+    bdlmt::TimerEventScheduler::Handle d_authenticationTimerHandle;
+
+    /// The mutex to protect the AuthenticationTimerHandle and
+    bslmt::Mutex d_timerHandleMutex;
+
     /// The mutex to protect the AuthenticationResult.
     mutable bslmt::Mutex d_mutex;
 
@@ -105,6 +112,8 @@ class AuthenticationContext {
     AuthenticationContext& setAuthenticationResult(
         const bsl::shared_ptr<mqbplug::AuthenticationResult>& value);
     AuthenticationContext&
+    setAuthenticationTimerHandle(bdlmt::TimerEventScheduler::Handle value);
+    AuthenticationContext&
     setInitialConnectionContext(InitialConnectionContext* value);
     AuthenticationContext&
     setAuthenticationMessage(const bmqp_ctrlmsg::AuthenticationMessage& value);
@@ -124,6 +133,12 @@ class AuthenticationContext {
     const bsl::shared_ptr<mqbplug::AuthenticationResult>&
     authenticationResult() const;
 
+    /// This function holds a mutex lock while accessing the
+    /// `d_authenticationTimerHandle` to ensure thread safety.
+    bdlmt::TimerEventScheduler::Handle authenticationTimerHandle() const;
+
+    bslmt::Mutex& authenticationTimerHandleMutex();
+
     InitialConnectionContext* initialConnectionContext() const;
     const bmqp_ctrlmsg::AuthenticationMessage& authenticationMessage() const;
     bmqp::EncodingType::Enum authenticationEncodingType() const;