use plugin to authenticate

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

Author: emelialei88

src/groups/bmq/bmqp/bmqp_event.h

@@ -161,6 +161,10 @@ class Event {
     /// returns true.
     Event clone(bslma::Allocator* allocator) const;
 
+    /// Return the encoding type of this Authentication event.  The
+    /// behavior is undefined unless `isAuthenticationEvent()` returns true.
+    EncodingType::Enum authenticationEventEncodingType() const;
+
     /// Return the type of this event.  The behavior is undefined unless
     /// `isValid()` returns true.
     EventType::Enum type() const;
@@ -336,6 +340,10 @@ int Event::loadSchemaEvent(TYPE* message) const
     if (d_header->type() == EventType::e_CONTROL) {
         encodingType = EventHeaderUtil::controlEventEncodingType(*d_header);
     }
+    else if (d_header->type() == EventType::e_AUTHENTICATION) {
+        encodingType = EventHeaderUtil::authenticationEventEncodingType(
+            *d_header);
+    }
 
     bmqu::MemOutStream os;
     int                rc = ProtocolUtil::decodeMessage(os,
@@ -428,6 +436,14 @@ inline Event Event::clone(bslma::Allocator* allocator) const
     return Event(d_blob_p, allocator, true /* clone == true */);
 }
 
+inline EncodingType::Enum Event::authenticationEventEncodingType() const
+{
+    // PRECONDITIONS
+    BSLS_ASSERT_SAFE(isAuthenticationEvent());
+
+    return EventHeaderUtil::authenticationEventEncodingType(*d_header);
+}
+
 inline EventType::Enum Event::type() const
 {
     // PRECONDITIONS

src/groups/bmq/bmqp/bmqp_protocol.h

@@ -928,10 +928,20 @@ struct EventHeaderUtil {
     static void setControlEventEncodingType(EventHeader*       eventHeader,
                                             EncodingType::Enum type);
 
+    /// Set the appropriate bits in the specified `eventHeader` to represent
+    /// the specified encoding `type` for an authentication event.
+    static void setAuthenticationEventEncodingType(EventHeader* eventHeader,
+                                                   EncodingType::Enum type);
+
     /// Return the encoding type for a control event represented by the
     /// appropriate bits in the specified `eventHeader`.
     static EncodingType::Enum
     controlEventEncodingType(const EventHeader& eventHeader);
+
+    /// Return the encoding type for a authentication event represented by the
+    /// appropriate bits in the specified `eventHeader`.
+    static EncodingType::Enum
+    authenticationEventEncodingType(const EventHeader& eventHeader);
 };
 
 // ===================
@@ -3857,6 +3867,26 @@ EventHeaderUtil::setControlEventEncodingType(EventHeader*       eventHeader,
     eventHeader->setTypeSpecific(typeSpecific);
 }
 
+inline void
+EventHeaderUtil::setAuthenticationEventEncodingType(EventHeader* eventHeader,
+                                                    EncodingType::Enum type)
+{
+    // PRECONDITIONS
+    BSLS_ASSERT_SAFE(eventHeader->type() == EventType::e_AUTHENTICATION);
+    BSLS_ASSERT_SAFE(type != EncodingType::e_UNKNOWN);
+
+    unsigned char typeSpecific = eventHeader->typeSpecific();
+
+    // Reset the bits for encoding type
+    typeSpecific &= static_cast<unsigned char>(~k_CONTROL_EVENT_ENCODING_MASK);
+
+    // Set those bits to represent 'type'
+    typeSpecific |= static_cast<unsigned char>(
+        type << k_CONTROL_EVENT_ENCODING_START_IDX);
+
+    eventHeader->setTypeSpecific(typeSpecific);
+}
+
 inline EncodingType::Enum
 EventHeaderUtil::controlEventEncodingType(const EventHeader& eventHeader)
 {
@@ -3869,6 +3899,18 @@ EventHeaderUtil::controlEventEncodingType(const EventHeader& eventHeader)
     return static_cast<EncodingType::Enum>(encodingType);
 }
 
+inline EncodingType::Enum EventHeaderUtil::authenticationEventEncodingType(
+    const EventHeader& eventHeader)
+{
+    // PRECONDITIONS
+    BSLS_ASSERT_SAFE(eventHeader.type() == EventType::e_AUTHENTICATION);
+
+    const unsigned char typeSpecific = eventHeader.typeSpecific();
+    const int encodingType = (typeSpecific & k_CONTROL_EVENT_ENCODING_MASK) >>
+                             k_CONTROL_EVENT_ENCODING_START_IDX;
+    return static_cast<EncodingType::Enum>(encodingType);
+}
+
 // -------------------
 // struct OptionHeader
 // -------------------

src/groups/bmq/bmqp/bmqp_schemaeventbuilder.h

@@ -250,6 +250,10 @@ int SchemaEventBuilder::setMessage(const TYPE& message, EventType::Enum type)
         EventHeaderUtil::setControlEventEncodingType(eventHeader,
                                                      d_encodingType);
     }
+    else if (type == EventType::e_AUTHENTICATION) {
+        EventHeaderUtil::setAuthenticationEventEncodingType(eventHeader,
+                                                            d_encodingType);
+    }
 
     // Append appropriate encoding of 'message' to the blob
     int rc = ProtocolUtil::encodeMessage(d_errorStream,

src/groups/mqb/mqba/mqba_application.cpp

@@ -332,8 +332,9 @@ 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")),
+        new (*d_allocator_p) Authenticator(d_authenticationController_mp.get(),
+                                           &d_blobSpPool,
+                                           d_allocators.get("Authenticator")),
         d_allocator_p);
 
     SessionNegotiator* sessionNegotiator = new (*d_allocator_p)
@@ -547,8 +548,8 @@ void Application::stop()
     STOP_OBJ(d_domainManager_mp, "DomainManager");
     STOP_OBJ(d_dispatcher_mp, "Dispatcher");
     STOP_OBJ(d_configProvider_mp, "ConfigProvider");
-    STOP_OBJ(d_statController_mp, "StatController");
     STOP_OBJ(d_authenticationController_mp, "AuthenticationController");
+    STOP_OBJ(d_statController_mp, "StatController");
     STOP_OBJ(d_pluginManager_mp, "PluginManager");
 
     // and now DESTROY everything
@@ -557,8 +558,8 @@ void Application::stop()
     DESTROY_OBJ(d_transportManager_mp, "TransportManager");
     DESTROY_OBJ(d_dispatcher_mp, "Dispatcher");
     DESTROY_OBJ(d_configProvider_mp, "ConfigProvider");
-    DESTROY_OBJ(d_statController_mp, "StatController");
     DESTROY_OBJ(d_authenticationController_mp, "AuthenticationController");
+    DESTROY_OBJ(d_statController_mp, "StatController");
     DESTROY_OBJ(d_pluginManager_mp, "PluginManager");
 
     BALL_LOG_INFO << "BMQbrkr stopped";

src/groups/mqb/mqba/mqba_authenticator.cpp

@@ -29,19 +29,24 @@
 #include <mqbblp_clustercatalog.h>
 #include <mqbnet_authenticationcontext.h>
 #include <mqbnet_initialconnectioncontext.h>
+#include <mqbplug_authenticator.h>
 
 // BMQ
 #include <bmqio_status.h>
 #include <bmqp_ctrlmsg_messages.h>
 #include <bmqp_protocol.h>
 #include <bmqp_schemaeventbuilder.h>
+#include <bmqsys_threadutil.h>
 #include <bmqu_memoutstream.h>
 
 // BDE
 #include <ball_log.h>
 #include <bdlma_sequentialallocator.h>
+#include <bdlmt_threadpool.h>
 #include <bsl_memory.h>
 #include <bsl_ostream.h>
+#include <bsls_nullptr.h>
+#include <bsls_timeinterval.h>
 
 namespace BloombergLP {
 namespace mqba {
@@ -62,6 +67,13 @@ int Authenticator::onAuthenticationRequest(
     const bmqp_ctrlmsg::AuthenticationMessage& authenticationMsg,
     const InitialConnectionContextSp&          context)
 {
+    enum RcEnum {
+        // Value for the various RC error categories
+        rc_SUCCESS                    = 0,
+        rc_AUTHENTICATING_IN_PROGRESS = -1,
+        rc_FAIL_TO_ENQUEUE_JOB        = -2,
+    };
+
     // PRECONDITIONS
     BSLS_ASSERT_SAFE(authenticationMsg.isAuthenticateRequestValue());
     BSLS_ASSERT_SAFE(context->isIncoming());
@@ -70,10 +82,6 @@ int Authenticator::onAuthenticationRequest(
                    << context->channel()->peerUri()
                    << "': " << authenticationMsg;
 
-    bmqp_ctrlmsg::AuthenticationMessage authenticationResponse;
-    bmqp_ctrlmsg::AuthenticateResponse& response =
-        authenticationResponse.makeAuthenticateResponse();
-
     // Create an AuthenticationContext for that connection
     bsl::shared_ptr<mqbnet::AuthenticationContext> authenticationContext =
         bsl::allocate_shared<mqbnet::AuthenticationContext>(
@@ -85,22 +93,18 @@ int Authenticator::onAuthenticationRequest(
         );
 
     context->setAuthenticationContext(authenticationContext);
+    authenticationContext->setInitialConnectionContext(context.get());
 
-    // Always succeeds for now
-    // TODO: For later implementation, plugins will perform authentication,
-    // taking the `AuthenticationContext` and updates it with the
-    // authentication result.
-    response.status().category() = bmqp_ctrlmsg::StatusCategory::E_SUCCESS;
-    response.status().code()     = 0;
-    response.lifetimeMs()        = 10 * 60 * 1000;
-
-    authenticationContext->state().testAndSwap(
-        mqbnet::AuthenticationContext::State::e_AUTHENTICATING,
-        mqbnet::AuthenticationContext::State::e_AUTHENTICATED);
+    // Authenticate
+    int rc = d_threadPool.enqueueJob(
+        bdlf::BindUtil::bind(&Authenticator::authenticate, this, context));
 
-    int rc = sendAuthenticationMessage(errorDescription,
-                                       authenticationResponse,
-                                       authenticationContext);
+    if (rc != 0) {
+        errorDescription << "Failed to enqueue authentication job for '"
+                         << context->channel()->peerUri() << "' [rc: " << rc
+                         << ", message: " << authenticationMsg << "]";
+        return rc_FAIL_TO_ENQUEUE_JOB;  // RETURN
+    }
 
     return rc;
 }
@@ -118,7 +122,7 @@ int Authenticator::onAuthenticationResponse(
 int Authenticator::sendAuthenticationMessage(
     bsl::ostream&                              errorDescription,
     const bmqp_ctrlmsg::AuthenticationMessage& message,
-    const AuthenticationContextSp&             context)
+    const InitialConnectionContextSp&          context)
 {
     enum RcEnum {
         // Value for the various RC error categories
@@ -127,12 +131,10 @@ int Authenticator::sendAuthenticationMessage(
         rc_WRITE_FAILURE = -2
     };
 
-    bmqp::EncodingType::Enum encodingType = bmqp::EncodingType::e_BER;
-
     bdlma::LocalSequentialAllocator<2048> localAllocator(d_allocator_p);
 
     bmqp::SchemaEventBuilder builder(d_blobSpPool_p,
-                                     encodingType,
+                                     context->authenticationEncodingType(),
                                      &localAllocator);
 
     int rc = builder.setMessage(message, bmqp::EventType::e_AUTHENTICATION);
@@ -144,8 +146,9 @@ int Authenticator::sendAuthenticationMessage(
 
     // Send response event
     bmqio::Status status;
-    context->initialConnectionContext()->channel()->write(&status,
-                                                          *builder.blob());
+
+    context->channel()->write(&status, *builder.blob());
+
     if (!status) {
         errorDescription << "Failed sending AuthenticationMessage "
                          << "[status: " << status << ", message: " << message
@@ -162,11 +165,104 @@ void Authenticator::initiateOutboundAuthentication(
     BALL_LOG_ERROR << "Not Implemented";
 }
 
+void Authenticator::authenticate(const InitialConnectionContextSp& context)
+{
+    // PRECONDITIONS
+    BSLS_ASSERT(context->authenticationContext());
+    BSLS_ASSERT(context->authenticationContext()->initialConnectionContext());
+
+    const bmqp_ctrlmsg::AuthenticateRequest& authenticateRequest =
+        context->authenticationContext()
+            ->authenticationMessage()
+            .authenticateRequest();
+
+    bsl::shared_ptr<mqbplug::AuthenticationResult> result;
+    mqbplug::AuthenticationData                    authenticationData(
+        authenticateRequest.data().value(),
+        context->channel()->peerUri());
+
+    bmqp_ctrlmsg::AuthenticationMessage authenticationResponse;
+    bmqp_ctrlmsg::AuthenticateResponse& response =
+        authenticationResponse.makeAuthenticateResponse();
+
+    bmqu::MemOutStream authenticationErrorStream;
+
+    BALL_LOG_INFO << "Authenticating connection '"
+                  << context->channel()->peerUri() << "' with mechanism '"
+                  << authenticateRequest.mechanism() << "'";
+
+    const int authn_rc = d_authnController_p->authenticate(
+        authenticationErrorStream,
+        &result,
+        authenticateRequest.mechanism(),
+        authenticationData);
+    if (authn_rc != 0) {
+        BALL_LOG_ERROR << "Authentication failed for connection '"
+                       << context->channel()->peerUri() << "' with mechanism '"
+                       << authenticateRequest.mechanism()
+                       << "' [rc: " << authn_rc
+                       << ", error: " << authenticationErrorStream.str()
+                       << "]";
+
+        response.status().code()     = authn_rc;
+        response.status().category() = bmqp_ctrlmsg::StatusCategory::E_REFUSED;
+        response.status().message()  = authenticationErrorStream.str();
+
+        context->setAuthenticationContext(bsl::nullptr_t());
+    }
+    else {
+        response.status().code()     = 0;
+        response.status().category() = bmqp_ctrlmsg::StatusCategory::E_SUCCESS;
+        response.lifetimeMs()        = result->lifetimeMs();
+
+        context->authenticationContext()->setAuthenticationResult(result);
+
+        context->authenticationContext()->state().testAndSwap(
+            State::e_AUTHENTICATING,
+            State::e_AUTHENTICATED);
+    }
+
+    bmqu::MemOutStream sendResponseErrorStream;
+
+    const int send_rc = sendAuthenticationMessage(sendResponseErrorStream,
+                                                  authenticationResponse,
+                                                  context);
+
+    // TODO: if we close channel here, for initial connection, we won't be able
+    // to logOpenSessionTime
+    // Maybe this shoule be a callback. So can be triggered under different
+    // senarios
+    if (authn_rc != 0) {
+        bmqio::Status status(bmqio::StatusCategory::e_GENERIC_ERROR,
+                             "AuthenticationError",
+                             authn_rc,
+                             d_allocator_p);
+        context->channel()->close(status);
+    }
+    else if (send_rc != 0) {
+        BALL_LOG_ERROR << sendResponseErrorStream.str();
+
+        bmqio::Status status(bmqio::StatusCategory::e_GENERIC_ERROR,
+                             "AuthenticationError",
+                             send_rc,
+                             d_allocator_p);
+        context->channel()->close(status);
+    }
+}
+
 // CREATORS
-Authenticator::Authenticator(BlobSpPool*       blobSpPool,
-                             bslma::Allocator* allocator)
-: d_allocator_p(allocator)
+Authenticator::Authenticator(
+    mqbauthn::AuthenticationController* authnController,
+    BlobSpPool*                         blobSpPool,
+    bslma::Allocator*                   allocator)
+: d_authnController_p(authnController)
+, d_threadPool(bmqsys::ThreadUtil::defaultAttributes(),
+               0,                                            // min threads
+               100,                                          // max threads
+               bsls::TimeInterval(120).totalMilliseconds(),  // idle time
+               allocator)
 , d_blobSpPool_p(blobSpPool)
+, d_allocator_p(allocator)
 {
     // NOTHING
 }
@@ -177,6 +273,23 @@ Authenticator::~Authenticator()
     // NOTHING: (required because of inheritance)
 }
 
+int Authenticator::start(bsl::ostream& errorDescription)
+{
+    int rc = d_threadPool.start();
+    if (rc != 0) {
+        errorDescription << "Failed to start thread pool for Authenticator"
+                         << "[rc: " << rc << "]";
+        return rc;  // RETURN
+    }
+
+    return 0;
+}
+
+void Authenticator::stop()
+{
+    d_threadPool.stop();
+}
+
 int Authenticator::handleAuthentication(
     bsl::ostream&                              errorDescription,
     bool*                                      isContinueRead,