added reauthentication

Signed-off-by: Emelia Lei <[email protected]>

Author: emelialei88

src/groups/mqb/mqba/mqba_application.cpp

@@ -359,14 +359,15 @@ int Application::start(bsl::ostream& errorDescription)
     bslma::ManagedPtr<mqbnet::InitialConnectionHandler>
         initialConnectionHandlerMp(
             new (*d_allocator_p) InitialConnectionHandler(
-                authenticatorMp,
                 negotiatorMp,
+                authenticatorMp.get(),
                 d_allocators.get("InitialConnectionHandler")),
             d_allocator_p);
 
     d_transportManager_mp.load(new (*d_allocator_p) mqbnet::TransportManager(
                                    d_scheduler_p,
                                    &d_bufferFactory,
+                                   authenticatorMp,
                                    initialConnectionHandlerMp,
                                    d_statController_mp.get(),
                                    d_allocators.get("TransportManager")),

src/groups/mqb/mqba/mqba_authenticator.cpp

@@ -144,14 +144,47 @@ class Authenticator : public mqbnet::Authenticator {
     int sendAuthenticationMessage(
         bsl::ostream&                              errorDescription,
         const bmqp_ctrlmsg::AuthenticationMessage& message,
-        const InitialConnectionContextSp&          context);
-
-    /// Authenticate using a plugin based on the mechanism specified in the
-    /// `AuthenticateRequest` message.  If the authentication fails, reset
-    /// `AuthenticationContext` and close the channel.  Send an
-    /// `AuthenticationResponse` message back to the peer with the status of
-    /// the authentication.
-    void authenticate(const InitialConnectionContextSp& context);
+        const bsl::shared_ptr<bmqio::Channel>&     channel,
+        bmqp::EncodingType::Enum                   authenticationEncodingType);
+
+    /// Schedule an authentication job in the thread pool using the
+    /// specified `context` and `channel`.  Return 0 on success, or a
+    /// non-zero error code and populate the specified `errorDescription`
+    /// with a description of the error otherwise.
+    int authenticateAsync(bsl::ostream&                  errorDescription,
+                          const AuthenticationContextSp& context,
+                          const bsl::shared_ptr<bmqio::Channel>& channel);
+
+    /// Schedule a re-authentication job in the thread pool using the
+    /// specified `context` and `channel`.  Return 0 on success, or a
+    /// non-zero error code and populate the specified `errorDescription`
+    /// with a description of the error otherwise.
+    int reAuthenticateAsync(bsl::ostream&                  errorDescription,
+                            const AuthenticationContextSp& context,
+                            const bsl::shared_ptr<bmqio::Channel>& channel);
+
+    /// Authenticate the connection using the `AuthenticationMessage` stored in
+    /// `context`.  If authentication fails, invoke
+    /// `initialConnectionCompleteCb` to close the `channel`. Also, update the
+    /// state of `context` as appropriate.
+    void authenticate(const AuthenticationContextSp&         context,
+                      const bsl::shared_ptr<bmqio::Channel>& channel);
+
+    /// Re-authenticate the connection using the `AuthenticationMessage`
+    /// stored in `context`.  If re-authentication fails, invoke
+    /// `initialConnectionCompleteCb` to close the `channel`. Also, update the
+    /// state of `context` as appropriate.
+    void reAuthenticate(const AuthenticationContextSp&         context,
+                        const bsl::shared_ptr<bmqio::Channel>& channel);
+
+    /// Handle re-authentication depending on the type of AuthenticationMessage
+    /// stored in `context` for the specified `context` and `channel`. If the
+    /// re-authentication is successful, return 0; otherwise, return a non-zero
+    /// error code and populate the specified `errorDescription` with a
+    /// description of the error.
+    int handleReauthentication(bsl::ostream&                  errorDescription,
+                               const AuthenticationContextSp& context,
+                               const bsl::shared_ptr<bmqio::Channel>& channel);
 
   public:
     // TRAITS

src/groups/mqb/mqba/mqba_authenticator.h

@@ -153,6 +153,7 @@
 #include <mqbcfg_brokerconfig.h>
 #include <mqbi_cluster.h>
 #include <mqbi_queue.h>
+#include <mqbnet_authenticationcontext.h>
 #include <mqbnet_tcpsessionfactory.h>
 #include <mqbstat_brokerstats.h>
 #include <mqbu_messageguidutil.h>
@@ -2631,14 +2632,16 @@ ClientSession::ClientSession(
     const bsl::shared_ptr<bmqio::Channel>&  channel,
     const bmqp_ctrlmsg::NegotiationMessage& negotiationMessage,
     const bsl::string&                      sessionDescription,
-    mqbi::Dispatcher*                       dispatcher,
-    mqbblp::ClusterCatalog*                 clusterCatalog,
-    mqbi::DomainFactory*                    domainFactory,
-    bslma::ManagedPtr<bmqst::StatContext>&  clientStatContext,
-    ClientSessionState::BlobSpPool*         blobSpPool,
-    bdlbb::BlobBufferFactory*               bufferFactory,
-    bdlmt::EventScheduler*                  scheduler,
-    bslma::Allocator*                       allocator)
+    const bsl::shared_ptr<mqbnet::AuthenticationContext>&
+                                           authenticationContext,
+    mqbi::Dispatcher*                      dispatcher,
+    mqbblp::ClusterCatalog*                clusterCatalog,
+    mqbi::DomainFactory*                   domainFactory,
+    bslma::ManagedPtr<bmqst::StatContext>& clientStatContext,
+    ClientSessionState::BlobSpPool*        blobSpPool,
+    bdlbb::BlobBufferFactory*              bufferFactory,
+    bdlmt::EventScheduler*                 scheduler,
+    bslma::Allocator*                      allocator)
 : d_self(this)  // use default allocator
 , d_operationState(e_RUNNING)
 , d_isDisconnecting(false)
@@ -2650,6 +2653,7 @@ ClientSession::ClientSession(
       bmqp::MessagePropertiesFeatures::k_MESSAGE_PROPERTIES_EX,
       d_clientIdentity_p->features()))
 , d_description(sessionDescription, allocator)
+, d_authenticationContext(authenticationContext)
 , d_channel_sp(channel)
 , d_state(clientStatContext,
           blobSpPool,
@@ -2716,7 +2720,56 @@ void ClientSession::processEvent(const bmqp::Event& event,
 {
     // executed by the *IO* thread
 
-    if (event.isControlEvent()) {
+    // TODO: uncomment when default credential support is implemented
+    // if (!event.isAuthenticationEvent() && !d_authenticationContext) {
+    //     BALL_LOG_ERROR << "The authentication lifetime has expired.  Need to
+    //     "
+    //                       "re-authenticate.";
+    //     return;  // RETURN
+    // }
+
+    if (event.isAuthenticationEvent()) {
+        if (d_authenticationContext->state().testAndSwap(
+                AuthnState::e_AUTHENTICATED,
+                AuthnState::e_AUTHENTICATING) != AuthnState::e_AUTHENTICATED) {
+            BALL_LOG_ERROR << "#CLIENT_IMPROPER_BEHAVIOR " << description()
+                           << ": received Authentication event while "
+                              "authentication is in progress";
+            return;  // RETURN
+        }
+
+        bmqp_ctrlmsg::AuthenticationMessage authenticationMessage;
+        int rc = event.loadAuthenticationEvent(&authenticationMessage);
+        if (rc != 0) {
+            BALL_LOG_ERROR << "#CORRUPTED_EVENT " << description()
+                           << ": Received invalid authentication message "
+                              "from client [reason: 'failed to decode', rc: "
+                           << rc << "]:\n"
+                           << bmqu::BlobStartHexDumper(event.blob());
+            return;  // RETURN
+        }
+
+        BALL_LOG_INFO << description() << ": Received authentication message: "
+                      << authenticationMessage;
+
+        d_authenticationContext->setAuthenticationMessage(
+            authenticationMessage);
+        d_authenticationContext->setAuthenticationEncodingType(
+            event.authenticationEventEncodingType());
+
+        bmqu::MemOutStream errorStream;
+        rc = d_authenticationContext->reAuthenticateCb()(
+            errorStream,
+            d_authenticationContext,
+            d_channel_sp);
+        if (rc != 0) {
+            BALL_LOG_ERROR << "#AUTHENTICATION_FAILED " << description()
+                           << ": Authentication failed [reason: '"
+                           << errorStream.str() << "', rc: " << rc << "]";
+            return;  // RETURN
+        }
+    }
+    else if (event.isControlEvent()) {
         bdlma::LocalSequentialAllocator<2048> localAllocator(
             d_state.d_allocator_p);
         bmqp_ctrlmsg::ControlMessage controlMessage(&localAllocator);

src/groups/mqb/mqba/mqba_clientsession.cpp

@@ -33,6 +33,8 @@
 #include <mqbi_dispatcher.h>
 #include <mqbi_domain.h>
 #include <mqbi_queue.h>
+#include <mqbnet_authenticationcontext.h>
+#include <mqbnet_initialconnectioncontext.h>
 #include <mqbnet_session.h>
 #include <mqbstat_queuestats.h>
 
@@ -255,6 +257,8 @@ class ClientSession : public mqbnet::Session,
 
     typedef bsl::function<void(void)> VoidFunctor;
 
+    typedef mqbnet::AuthenticationContext::State AuthnState;
+
     /// Enum to signify the session's operation state.
     enum OperationState {
         /// Running normally.
@@ -337,6 +341,11 @@ class ClientSession : public mqbnet::Session,
     /// Short identifier for this session.
     bsl::string d_description;
 
+    /// The authenticationContext first created during authentication in
+    /// initial connection, and later on may get updated during
+    /// re-authentication.
+    bsl::shared_ptr<mqbnet::AuthenticationContext> d_authenticationContext;
+
     /// Channel associated with this session.
     bsl::shared_ptr<bmqio::Channel> d_channel_sp;
 
@@ -628,14 +637,16 @@ class ClientSession : public mqbnet::Session,
     ClientSession(const bsl::shared_ptr<bmqio::Channel>&  channel,
                   const bmqp_ctrlmsg::NegotiationMessage& negotiationMessage,
                   const bsl::string&                      sessionDescription,
-                  mqbi::Dispatcher*                       dispatcher,
-                  mqbblp::ClusterCatalog*                 clusterCatalog,
-                  mqbi::DomainFactory*                    domainFactory,
-                  bslma::ManagedPtr<bmqst::StatContext>&  clientStatContext,
-                  ClientSessionState::BlobSpPool*         blobSpPool,
-                  bdlbb::BlobBufferFactory*               bufferFactory,
-                  bdlmt::EventScheduler*                  scheduler,
-                  bslma::Allocator*                       allocator);
+                  const bsl::shared_ptr<mqbnet::AuthenticationContext>&
+                                                         authenticationContext,
+                  mqbi::Dispatcher*                      dispatcher,
+                  mqbblp::ClusterCatalog*                clusterCatalog,
+                  mqbi::DomainFactory*                   domainFactory,
+                  bslma::ManagedPtr<bmqst::StatContext>& clientStatContext,
+                  ClientSessionState::BlobSpPool*        blobSpPool,
+                  bdlbb::BlobBufferFactory*              bufferFactory,
+                  bdlmt::EventScheduler*                 scheduler,
+                  bslma::Allocator*                      allocator);
 
     /// Destructor
     ~ClientSession() BSLS_KEYWORD_OVERRIDE;

src/groups/mqb/mqba/mqba_clientsession.h

@@ -26,6 +26,7 @@
 #include <mqbmock_queue.h>
 #include <mqbmock_queueengine.h>
 #include <mqbmock_queuehandle.h>
+#include <mqbnet_initialconnectioncontext.h>
 #include <mqbstat_brokerstats.h>
 #include <mqbstat_queuestats.h>
 #include <mqbu_messageguidutil.h>
@@ -689,6 +690,7 @@ class TestBench {
     , d_cs(d_channel,
            negotiationMessage,
            "sessionDescription",
+           bsl::shared_ptr<mqbnet::AuthenticationContext>(),
            setInDispatcherThread(&d_mockDispatcher),
            0,  // ClusterCatalog
            &d_mockDomainFactory,
@@ -2125,7 +2127,7 @@ static void test11_initiateShutdown()
         100000000);  // 100 ms
     bslmt::TimedSemaphore      semaphore;
     bmqp::Protocol::MsgGroupId msgGroupId(bmqtst::TestHelperUtil::allocator());
-    const unsigned int             subscriptionId =
+    const unsigned int         subscriptionId =
         bmqp::Protocol::k_DEFAULT_SUBSCRIPTION_ID;
     const unsigned int subQueueId = bmqp::QueueId::k_DEFAULT_SUBQUEUE_ID;
 
@@ -2376,11 +2378,11 @@ static void test11_initiateShutdown()
 
     PV("Confirm multiple messsages while shutting down");
     {
-        const int       NUM_MESSAGES = 5;
+        const int                      NUM_MESSAGES = 5;
         TestBench                      tb(client(e_FirstHop),
                      isAtMostOnce,
                      bmqtst::TestHelperUtil::allocator());
-        bsls::AtomicInt callbackCounter(0);
+        bsls::AtomicInt                callbackCounter(0);
         bsl::vector<bmqt::MessageGUID> guids(
             bmqtst::TestHelperUtil::allocator());
         guids.reserve(NUM_MESSAGES);

src/groups/mqb/mqba/mqba_clientsession.t.cpp

@@ -211,11 +211,10 @@ int InitialConnectionHandler::processBlob(
         context->negotiationContext()->d_negotiationMessage =
             bsl::get<bmqp_ctrlmsg::NegotiationMessage>(message.value());
 
-        rc = d_negotiator_mp->createSessionOnMsgType(
-            errorDescription,
-            session,
-            isContinueRead,
-            context->negotiationContext());
+        rc = d_negotiator_mp->createSessionOnMsgType(errorDescription,
+                                                     session,
+                                                     isContinueRead,
+                                                     context);
     }
 
     return rc;
@@ -334,9 +333,9 @@ void InitialConnectionHandler::complete(
 }
 
 InitialConnectionHandler::InitialConnectionHandler(
-    bslma::ManagedPtr<mqbnet::Authenticator>& authenticator,
-    bslma::ManagedPtr<mqbnet::Negotiator>&    negotiator,
-    bslma::Allocator*                         allocator)
+    bslma::ManagedPtr<mqbnet::Negotiator>& negotiator,
+    mqbnet::Authenticator*                 authenticator,
+    bslma::Allocator*                      allocator)
 : d_authenticator_mp(authenticator)
 , d_negotiator_mp(negotiator)
 , d_allocator_p(allocator)
@@ -397,8 +396,10 @@ void InitialConnectionHandler::handleConnectionFlow(
         rc = scheduleRead(errStream, context);
     }
     else {
-        rc = d_negotiator_mp->negotiateOutbound(errStream,
-                                                context->negotiationContext());
+        // TODO: When we are ready to move on to the next step, we should
+        // call `authenticationOutbound` here instead before calling
+        // `negotiateOutbound`.
+        rc = d_negotiator_mp->negotiateOutbound(errStream, context);
 
         // Send outbound request success, continue to read
         if (rc == 0) {
@@ -414,17 +415,6 @@ void InitialConnectionHandler::handleConnectionFlow(
     guard.release();
 }
 
-int InitialConnectionHandler::start(bsl::ostream& errorDescription)
-{
-    int rc = d_authenticator_mp->start(errorDescription);
-    return rc;
-}
-
-void InitialConnectionHandler::stop()
-{
-    d_authenticator_mp->stop();
-}
-
 void InitialConnectionHandler::handleInitialConnection(
     const InitialConnectionContextSp& context)
 {

src/groups/mqb/mqba/mqba_initialconnectionhandler.cpp

@@ -71,7 +71,7 @@ class InitialConnectionHandler : public mqbnet::InitialConnectionHandler {
     // DATA
 
     /// Authenticator to use for authenticating a connection.
-    bslma::ManagedPtr<mqbnet::Authenticator> d_authenticator_mp;
+    mqbnet::Authenticator* d_authenticator_mp;
 
     /// Negotiator to use for converting a Channel to a Session.
     bslma::ManagedPtr<mqbnet::Negotiator> d_negotiator_mp;
@@ -148,24 +148,15 @@ class InitialConnectionHandler : public mqbnet::InitialConnectionHandler {
   public:
     // CREATORS
 
-    InitialConnectionHandler(
-        bslma::ManagedPtr<mqbnet::Authenticator>& authenticator,
-        bslma::ManagedPtr<mqbnet::Negotiator>&    negotiator,
-        bslma::Allocator*                         allocator);
+    InitialConnectionHandler(bslma::ManagedPtr<mqbnet::Negotiator>& negotiator,
+                             mqbnet::Authenticator* authenticator,
+                             bslma::Allocator*      allocator);
 
     /// Destructor
     ~InitialConnectionHandler() BSLS_KEYWORD_OVERRIDE;
 
     // MANIPULATORS
 
-    /// Start the InitialConnectionHandler.  Return 0 on success, or a
-    /// non-zero error code and populate the specified `errorDescription`
-    /// with a description of the error otherwise.
-    int start(bsl::ostream& errorDescription) BSLS_KEYWORD_OVERRIDE;
-
-    /// Stop the InitialConnectionHandler.
-    void stop() BSLS_KEYWORD_OVERRIDE;
-
     /// Method invoked by the client of this object to negotiate a session.
     /// The specified `context` is an in-out member holding the initial
     /// connection context to use, including an