Modify authenticator and initialconnectionhandler to handle authentication

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

Author: emelialei88

src/groups/bmq/bmqimp/README.md

@@ -13,13 +13,13 @@ its components directly, but rather use the accessor public APIs from 'bmqa'.
 
 Component Synopsis
 ------------------
-Component                         | Provides ...
-----------------------------------|-----------------------------------------------------------
-`bmqimp_application`              | the top level object to manipulate a session with bmqbrkr.
-`bmqimp_brokersession`            | the implementation of a session with the bmqbrkr.
-`bmqimp_event`                    | a value-semantic type representing an event.
-`bmqimp_eventqueue`               | a thread safe queue of pooled bmqimp::Event items.
-`bmqpimp_eventsstats`             | a mechanism to keep track of Events statistics.
-`bmqpimp_queue`                   | a type object to represent information about a queue.
-`bmqimp_stat`                     | utilities for stat manipulation.
-`bmqimp_initialconnectionchannelfactory` | a channel factory that negotiates with a peer.
+Component                                | Provides ...
+-----------------------------------------|-----------------------------------------------------------
+`bmqimp_application`                     | the top level object to manipulate a session with bmqbrkr.
+`bmqimp_brokersession`                   | the implementation of a session with the bmqbrkr.
+`bmqimp_event`                           | a value-semantic type representing an event.
+`bmqimp_eventqueue`                      | a thread safe queue of pooled bmqimp::Event items.
+`bmqpimp_eventsstats`                    | a mechanism to keep track of Events statistics.
+`bmqpimp_queue`                          | a type object to represent information about a queue.
+`bmqimp_stat`                            | utilities for stat manipulation.
+`bmqimp_initialconnectionchannelfactory` | a channel factory that authenticates a peer and and negotiates with it.

src/groups/bmq/bmqimp/bmqimp_brokersession.cpp

@@ -3621,8 +3621,7 @@ void BrokerSession::processPushEvent(const bmqp::Event& event)
                                                d_allocator_p);
         if (BSLS_PERFORMANCEHINT_PREDICT_UNLIKELY(rc != 0)) {
             BSLS_PERFORMANCEHINT_UNLIKELY_HINT;
-            BALL_LOG_ERROR << "Unable to flatten PUSH event"
-                           << " [rc: " << rc
+            BALL_LOG_ERROR << "Unable to flatten PUSH event [rc: " << rc
                            << ", length: " << event.blob()->length()
                            << ", eventMessageCount: " << eventMessageCount
                            << "]" << bsl::endl

src/groups/mqb/mqba/mqba_application.cpp

@@ -18,6 +18,7 @@
 
 #include <mqbscm_version.h>
 // MQB
+#include <mqba_authenticator.h>
 #include <mqba_configprovider.h>
 #include <mqba_dispatcher.h>
 #include <mqba_domainmanager.h>
@@ -316,6 +317,11 @@ int Application::start(bsl::ostream& errorDescription)
     }
 
     // Start the transport manager
+    bslma::ManagedPtr<mqbnet::Authenticator> authenticatorMp(
+        new (*d_allocator_p)
+            Authenticator(&d_blobSpPool, d_allocators.get("Authenticator")),
+        d_allocator_p);
+
     SessionNegotiator* sessionNegotiator = new (*d_allocator_p)
         SessionNegotiator(&d_bufferFactory,
                           d_dispatcher_mp.get(),
@@ -339,6 +345,7 @@ int Application::start(bsl::ostream& errorDescription)
     bslma::ManagedPtr<mqbnet::InitialConnectionHandler>
         initialConnectionHandlerMp(
             new (*d_allocator_p) InitialConnectionHandler(
+                authenticatorMp,
                 negotiatorMp,
                 d_allocators.get("InitialConnectionHandler")),
             d_allocator_p);

src/groups/mqb/mqba/mqba_authenticator.cpp

@@ -20,6 +20,7 @@
 
 /// Implementation Notes
 ///====================
+/// TODO
 
 // MQB
 #include <mqbblp_clustercatalog.h>
@@ -82,8 +83,6 @@ int Authenticator::onAuthenticationRequest(
         response.lifetimeMs()        = 10 * 60 * 1000;
     }
 
-    BALL_LOG_INFO << "send authn response " << authenticationResponse;
-
     int rc = sendAuthenticationMessage(errorDescription,
                                        authenticationResponse,
                                        context);
@@ -95,7 +94,9 @@ int Authenticator::onAuthenticationResponse(
     bsl::ostream&                  errorDescription,
     const AuthenticationContextSp& context)
 {
-    return 0;
+    BALL_LOG_ERROR << "Not Implemented";
+
+    return -1;
 }
 
 int Authenticator::sendAuthenticationMessage(
@@ -162,6 +163,8 @@ Authenticator::~Authenticator()
 }
 
 int Authenticator::handleAuthenticationOnMsgType(
+    bsl::ostream&                  errorDescription,
+    bool*                          isContinueRead,
     const AuthenticationContextSp& context)
 {
     enum RcEnum {
@@ -176,33 +179,22 @@ int Authenticator::handleAuthenticationOnMsgType(
     switch (context->d_authenticationMessage.selectionId()) {
     case bmqp_ctrlmsg::AuthenticationMessage::
         SELECTION_ID_AUTHENTICATE_REQUEST: {
-        BALL_LOG_INFO << "Received authn request: "
-                      << context->d_authenticationMessage;
         rc = onAuthenticationRequest(errStream, context);
     } break;  // BREAK
     case bmqp_ctrlmsg::AuthenticationMessage::
         SELECTION_ID_AUTHENTICATE_RESPONSE: {
-        BALL_LOG_INFO << "Received authn response: "
-                      << context->d_authenticationMessage;
+        rc = onAuthenticationResponse(errStream, context);
     } break;  // BREAK
     default: {
-        errStream << "Invalid authentication message received (unknown type): "
-                  << context->d_authenticationMessage;
-        bsl::string error(errStream.str().data(), errStream.str().length());
-        context->d_initialConnectionContext_p->initialConnectionCompleteCb()(
-            rc_ERROR,
-            error,
-            bsl::shared_ptr<mqbnet::Session>());
+        errorDescription
+            << "Invalid authentication message received (unknown type): "
+            << context->d_authenticationMessage;
         return rc_ERROR;  // RETURN
     }
     }
 
-    if (rc != rc_SUCCESS) {
-        bsl::string error(errStream.str().data(), errStream.str().length());
-        context->d_initialConnectionContext_p->initialConnectionCompleteCb()(
-            rc_ERROR,
-            error,
-            bsl::shared_ptr<mqbnet::Session>());
+    if (rc == rc_SUCCESS) {
+        *isContinueRead = true;
     }
 
     return rc;

src/groups/mqb/mqba/mqba_authenticator.h

@@ -154,7 +154,9 @@ class Authenticator : public mqbnet::Authenticator {
     // MANIPULATORS
     //   (virtual: mqbnet::Authenticator)
 
-    int handleAuthenticationOnMsgType(const AuthenticationContextSp& context)
+    int handleAuthenticationOnMsgType(bsl::ostream& errorDescription,
+                                      bool*         isContinueRead,
+                                      const AuthenticationContextSp& context)
         BSLS_KEYWORD_OVERRIDE;
 
     /// Send out outbound authentication message or reverse connection request

src/groups/mqb/mqba/mqba_initialconnectionhandler.cpp

@@ -175,17 +175,29 @@ int InitialConnectionHandler::processBlob(
         rc_INVALID_NEGOTIATION_MESSAGE = -1,
     };
 
-    bsl::optional<bmqp_ctrlmsg::NegotiationMessage> negotiationMsg;
+    bsl::optional<bmqp_ctrlmsg::AuthenticationMessage> authenticationMsg;
+    bsl::optional<bmqp_ctrlmsg::NegotiationMessage>    negotiationMsg;
 
     int rc = decodeInitialConnectionMessage(errorDescription,
                                             blob,
+                                            &authenticationMsg,
                                             &negotiationMsg);
 
-    if (rc != 0) {
+    if (rc != rc_SUCCESS) {
         return (rc * 10) + rc_INVALID_NEGOTIATION_MESSAGE;  // RETURN
     }
 
-    if (negotiationMsg.has_value()) {
+    // Authentication or Negotiation based on the type of message received.
+    if (authenticationMsg.has_value()) {
+        context->authenticationContext()->d_authenticationMessage =
+            authenticationMsg.value();
+
+        rc = d_authenticator_mp->handleAuthenticationOnMsgType(
+            errorDescription,
+            isContinueRead,
+            context->authenticationContext());
+    }
+    else if (negotiationMsg.has_value()) {
         context->negotiationContext()->d_negotiationMessage =
             negotiationMsg.value();
 
@@ -206,18 +218,21 @@ int InitialConnectionHandler::processBlob(
 }
 
 int InitialConnectionHandler::decodeInitialConnectionMessage(
-    bsl::ostream&                                    errorDescription,
-    const bdlbb::Blob&                               blob,
-    bsl::optional<bmqp_ctrlmsg::NegotiationMessage>* message)
+    bsl::ostream&                                       errorDescription,
+    const bdlbb::Blob&                                  blob,
+    bsl::optional<bmqp_ctrlmsg::AuthenticationMessage>* authenticationMsg,
+    bsl::optional<bmqp_ctrlmsg::NegotiationMessage>*    negotiationMsg)
 {
-    BSLS_ASSERT(message);
+    BSLS_ASSERT(authenticationMsg);
+    BSLS_ASSERT(negotiationMsg);
 
     enum RcEnum {
         // Value for the various RC error categories
-        rc_SUCCESS               = 0,
-        rc_INVALID_MESSAGE       = -1,
-        rc_NOT_CONTROL_EVENT     = -2,
-        rc_INVALID_CONTROL_EVENT = -3
+        rc_SUCCESS                      = 0,
+        rc_INVALID_MESSAGE              = -1,
+        rc_INVALID_EVENT                = -2,
+        rc_INVALID_AUTHENTICATION_EVENT = -3,
+        rc_INVALID_CONTROL_EVENT        = -4
     };
 
     bdlma::LocalSequentialAllocator<2048> localAllocator(d_allocator_p);
@@ -231,25 +246,41 @@ int InitialConnectionHandler::decodeInitialConnectionMessage(
         return rc_INVALID_MESSAGE;  // RETURN
     }
 
-    if (!event.isControlEvent()) {
-        errorDescription << "Invalid negotiation message received "
-                         << "(packet is not a ControlEvent):\n"
-                         << bmqu::BlobStartHexDumper(&blob);
-        return rc_NOT_CONTROL_EVENT;  // RETURN
-    }
-
-    bmqp_ctrlmsg::NegotiationMessage negotiationMessage;
-
-    int rc = event.loadControlEvent(&negotiationMessage);
+    bmqp_ctrlmsg::AuthenticationMessage authenticaionMessage;
+    bmqp_ctrlmsg::NegotiationMessage    negotiationMessage;
+
+    if (event.isAuthenticationEvent()) {
+        const int rc = event.loadAuthenticationEvent(&authenticaionMessage);
+        if (rc != 0) {
+            BALL_LOG_ERROR
+                << "Invalid response from broker [reason: 'authentication "
+                   "event is not an AuthenticationMessage', rc: "
+                << rc << "]: " << event;
+            return rc_INVALID_AUTHENTICATION_EVENT;  // RETURN
+        }
 
-    if (rc != 0) {
-        errorDescription << "Invalid negotiation message received (failed "
-                         << "decoding ControlEvent): [rc: " << rc << "]:\n"
-                         << bmqu::BlobStartHexDumper(&blob);
-        return rc_INVALID_CONTROL_EVENT;  // RETURN
+        *authenticationMsg = authenticaionMessage;
     }
+    else if (event.isControlEvent()) {
+        const int rc = event.loadControlEvent(&negotiationMessage);
+        if (rc != 0) {
+            BALL_LOG_ERROR
+                << "Invalid response from broker [reason: 'authentication "
+                   "event is not an AuthenticationMessage', rc: "
+                << rc << "]: " << event;
+
+            return rc_INVALID_CONTROL_EVENT;  // RETURN
+        }
 
-    *message = negotiationMessage;
+        *negotiationMsg = negotiationMessage;
+    }
+    else {
+        errorDescription
+            << "Invalid initial connection message received "
+            << "(packet is not an AuthenticationEvent or ControlEvent):\n"
+            << bmqu::BlobStartHexDumper(&blob);
+        return rc_INVALID_EVENT;  // RETURN
+    }
 
     return rc_SUCCESS;
 }
@@ -297,9 +328,11 @@ void InitialConnectionHandler::complete(
 }
 
 InitialConnectionHandler::InitialConnectionHandler(
-    bslma::ManagedPtr<mqbnet::Negotiator>& negotiator,
-    bslma::Allocator*                      allocator)
-: d_negotiator_mp(negotiator)
+    bslma::ManagedPtr<mqbnet::Authenticator>& authenticator,
+    bslma::ManagedPtr<mqbnet::Negotiator>&    negotiator,
+    bslma::Allocator*                         allocator)
+: d_authenticator_mp(authenticator)
+, d_negotiator_mp(negotiator)
 , d_allocator_p(allocator)
 {
 }
@@ -308,9 +341,21 @@ InitialConnectionHandler::~InitialConnectionHandler()
 {
 }
 
-void InitialConnectionHandler::handleInitialConnection(
+void InitialConnectionHandler::setupContext(
     const InitialConnectionContextSp& context)
 {
+    // Create an AuthenticationContext for that connection
+    bsl::shared_ptr<mqbnet::AuthenticationContext> authenticationContext;
+    authenticationContext.createInplace(d_allocator_p);
+
+    authenticationContext->d_initialConnectionContext_p = context.get();
+    authenticationContext->d_isReversed                 = false;
+    authenticationContext->d_clusterName                = "";
+    authenticationContext->d_connectionType =
+        mqbnet::ConnectionType::e_UNKNOWN;
+
+    context->setAuthenticationContext(authenticationContext);
+
     // Create an NegotiationContext for that connection
     bsl::shared_ptr<mqbnet::NegotiationContext> negotiationContext;
     negotiationContext.createInplace(d_allocator_p);
@@ -321,7 +366,11 @@ void InitialConnectionHandler::handleInitialConnection(
     negotiationContext->d_connectionType = mqbnet::ConnectionType::e_UNKNOWN;
 
     context->setNegotiationContext(negotiationContext);
+}
 
+void InitialConnectionHandler::handleConnectionFlow(
+    const InitialConnectionContextSp& context)
+{
     // Reading for inbound request or continue to read
     // after sending a request ourselves
 
@@ -363,5 +412,12 @@ void InitialConnectionHandler::handleInitialConnection(
     guard.release();
 }
 
+void InitialConnectionHandler::handleInitialConnection(
+    const InitialConnectionContextSp& context)
+{
+    setupContext(context);
+    handleConnectionFlow(context);
+}
+
 }  // close package namespace
 }  // close enterprise namespace

src/groups/mqb/mqba/mqba_initialconnectionhandler.h

@@ -20,6 +20,7 @@
 #include <mqbnet_initialconnectionhandler.h>
 
 // MQB
+#include <mqbnet_authenticator.h>
 #include <mqbnet_initialconnectioncontext.h>
 #include <mqbnet_negotiator.h>
 
@@ -39,6 +40,7 @@ namespace mqba {
 
 // FORWARD DECLARATION
 class SessionNegotiator;
+class Authenticator;
 
 // ==============================
 // class InitialConnectionHandler
@@ -53,7 +55,10 @@ class InitialConnectionHandler : public mqbnet::InitialConnectionHandler {
   private:
     // DATA
 
-    /// Negotiator to use for converting a Channel to a Session
+    /// Authenticator to use for authenticating a connection.
+    bslma::ManagedPtr<mqbnet::Authenticator> d_authenticator_mp;
+
+    /// Negotiator to use for converting a Channel to a Session.
     bslma::ManagedPtr<mqbnet::Negotiator> d_negotiator_mp;
 
     /// Allocator to use.
@@ -99,9 +104,10 @@ class InitialConnectionHandler : public mqbnet::InitialConnectionHandler {
     /// populate the specified `errorDescription` with a description of the
     /// error.
     int decodeInitialConnectionMessage(
-        bsl::ostream&                                    errorDescription,
-        const bdlbb::Blob&                               blob,
-        bsl::optional<bmqp_ctrlmsg::NegotiationMessage>* negotiationMsg);
+        bsl::ostream&                                       errorDescription,
+        const bdlbb::Blob&                                  blob,
+        bsl::optional<bmqp_ctrlmsg::AuthenticationMessage>* authenticationMsg,
+        bsl::optional<bmqp_ctrlmsg::NegotiationMessage>*    negotiationMsg);
 
     /// Schedule a read for the initial connection of the session of the
     /// specified `context`.  Return a non-zero code on error and
@@ -118,11 +124,17 @@ class InitialConnectionHandler : public mqbnet::InitialConnectionHandler {
                          const bsl::string&                      error,
                          const bsl::shared_ptr<mqbnet::Session>& session);
 
+    void setupContext(const InitialConnectionContextSp& context);
+
+    void handleConnectionFlow(const InitialConnectionContextSp& context);
+
   public:
     // CREATORS
 
-    InitialConnectionHandler(bslma::ManagedPtr<mqbnet::Negotiator>& negotiator,
-                             bslma::Allocator*                      allocator);
+    InitialConnectionHandler(
+        bslma::ManagedPtr<mqbnet::Authenticator>& authenticator,
+        bslma::ManagedPtr<mqbnet::Negotiator>&    negotiator,
+        bslma::Allocator*                         allocator);
 
     /// Destructor
     ~InitialConnectionHandler() BSLS_KEYWORD_OVERRIDE;