Skip to content

WIP Feat[MQB]: add authentication with basic logic #696

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
wants to merge 9 commits into
base: main
Choose a base branch
from
Draft

WIP Feat[MQB]: add authentication with basic logic #696

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
fe27c7b
generated bmqp_ctrl_message but are not formatted to Doxygen
emelialei88 Mar 21, 2025
74e138c
create classes for authentication
emelialei88 Mar 28, 2025
4b47a41
take authn out to be a new ctrlmsg type
emelialei88 Apr 10, 2025
7f8652e
modity authenticator (can't compile)
emelialei88 Apr 11, 2025
867f11b
add Authn Event Type
emelialei88 Apr 23, 2025
1d94c30
BMQ: change client side
emelialei88 Apr 23, 2025
338d9a9
modify authenticator
emelialei88 Apr 23, 2025
cef4a2d
Modify authenticator and initialconnectionhandler to handle authentic…
emelialei88 Apr 23, 2025
f750a50
minor fix
emelialei88 May 9, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions src/groups/bmq/bmqa/bmqa_session.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -246,6 +246,14 @@ int SessionUtil::createApplication(SessionImpl* sessionImpl)

static bsls::AtomicInt s_sessionInstanceCount(0);

// Authentication Message
bmqp_ctrlmsg::AuthenticationMessage authenticationMessage;
bmqp_ctrlmsg::AuthenticateRequest& ar =
authenticationMessage.makeAuthenticateRequest();
bsl::string str = "username:password";
ar.mechanism() = "basic";
ar.data() = bsl::vector<char>(str.begin(), str.end()); // hexBinary
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

A few notes:

  1. Binary hex introduces +100% protocol overhead to data.size() after encoding.
    If we use base64, we can have +33% protocol overhead.
    Probably it's not a problem, because we don't intend to send authn requests too often and don't want to send too much data here.

  2. Should binary hex be enforced? If so, it's worth to add checks in both client library (before sending a request) and on the broker side (when request is received).
    This also means that "username:password" data provided here will be rejected.

  3. Another way to enforce binary hex is to hide it from the library user.
    Let the user to just assign bsl::vector<char> data to anything, and convert it to binary hex when message is packed.

  4. What variations of capital letters does this hex binary support?
    abcdef, ABCDEF, or both abcdefABCDEF?


// Negotiation Message
bmqp_ctrlmsg::NegotiationMessage negotiationMessage;
bmqp_ctrlmsg::ClientIdentity& ci = negotiationMessage.makeClientIdentity();
Expand Down Expand Up @@ -311,6 +319,7 @@ int SessionUtil::createApplication(SessionImpl* sessionImpl)
sessionImpl->d_application_mp.load(
new (*(sessionImpl->d_allocator_p))
bmqimp::Application(options,
authenticationMessage,
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In its proper form, this probably won't be a concrete object, but rather something user-provided via SessionOptions that provides a callback to generate the raw information needed to construct this message.

negotiationMessage,
eventHandler,
sessionImpl->d_allocator_p),
Expand Down
20 changes: 10 additions & 10 deletions src/groups/bmq/bmqimp/README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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_negotiatedchannelfactory` | 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 negotiates with it.
41 changes: 23 additions & 18 deletions src/groups/bmq/bmqimp/bmqimp_application.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
// BMQ
#include <bmqex_executionpolicy.h>
#include <bmqex_systemexecutor.h>
#include <bmqimp_initialconnectionchannelfactory.h>
#include <bmqio_channelutil.h>
#include <bmqio_connectoptions.h>
#include <bmqio_status.h>
Expand Down Expand Up @@ -73,7 +74,7 @@ namespace {
// CONSTANTS
const double k_RECONNECT_INTERVAL_MS = 500;
const int k_RECONNECT_COUNT = bsl::numeric_limits<int>::max();
const bsls::Types::Int64 k_CHANNEL_LOW_WATERMARK = 512 * 1024;
const bsls::Types::Int64 k_CHANNEL_LOW_WATERMARK = 512 * 1024;
const int k_DEFAULT_MAX_MISSED_HEARTBEATS = 10;
const int k_DEFAULT_HEARTBEAT_INTERVAL_MS = 1000;

Expand Down Expand Up @@ -385,7 +386,7 @@ bmqt::GenericResult::Enum Application::startChannel()
.setAttemptInterval(attemptInterval)
.setAutoReconnect(true);

d_negotiatedChannelFactory.connect(
d_initialConnectionChannelFactory.connect(
&status,
&d_connectHandle_mp,
options,
Expand Down Expand Up @@ -550,10 +551,11 @@ Application::stateCb(bmqimp::BrokerSession::State::Enum oldState,
}

Application::Application(
const bmqt::SessionOptions& sessionOptions,
const bmqp_ctrlmsg::NegotiationMessage& negotiationMessage,
const EventQueue::EventHandlerCallback& eventHandlerCB,
bslma::Allocator* allocator)
const bmqt::SessionOptions& sessionOptions,
const bmqp_ctrlmsg::AuthenticationMessage& authenticationMessage,
const bmqp_ctrlmsg::NegotiationMessage& negotiationMessage,
const EventQueue::EventHandlerCallback& eventHandlerCB,
bslma::Allocator* allocator)
: d_allocatorStatContext(bmqst::StatContextConfiguration("Allocators",
allocator),
allocator)
Expand Down Expand Up @@ -597,12 +599,15 @@ Application::Application(
bdlf::PlaceHolders::_2), // handle
allocator),
allocator)
, d_negotiatedChannelFactory(
NegotiatedChannelFactoryConfig(&d_statChannelFactory,
negotiationMessage,
sessionOptions.connectTimeout(),
d_blobSpPool_sp.get(),
allocator),
, d_initialConnectionChannelFactory(
InitialConnectionChannelFactoryConfig(
&d_statChannelFactory,
authenticationMessage,
sessionOptions.connectTimeout(), // TODO: different for authn?
Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should we have a different timeout for authentication?

negotiationMessage,
sessionOptions.connectTimeout(),
d_blobSpPool_sp.get(),
allocator),
allocator)
, d_connectHandle_mp()
, d_brokerSession(&d_scheduler,
Expand Down Expand Up @@ -761,9 +766,9 @@ Application::createMonitor(const bsl::shared_ptr<bmqio::Channel>& channel)
{
int maxMissedHeartbeats = k_DEFAULT_MAX_MISSED_HEARTBEATS;

channel->properties().load(
&maxMissedHeartbeats,
NegotiatedChannelFactory::k_CHANNEL_PROPERTY_MAX_MISSED_HEARTBEATS);
channel->properties().load(&maxMissedHeartbeats,
InitialConnectionChannelFactory::
k_CHANNEL_PROPERTY_MAX_MISSED_HEARTBEATS);

bsl::shared_ptr<bmqp::HeartbeatMonitor> monitor(
new (d_allocator) bmqp::HeartbeatMonitor(maxMissedHeartbeats),
Expand All @@ -784,9 +789,9 @@ void Application::startHeartbeat(

int heartbeatIntervalMs = k_DEFAULT_HEARTBEAT_INTERVAL_MS;

channel->properties().load(
&heartbeatIntervalMs,
NegotiatedChannelFactory::k_CHANNEL_PROPERTY_HEARTBEAT_INTERVAL_MS);
channel->properties().load(&heartbeatIntervalMs,
InitialConnectionChannelFactory::
k_CHANNEL_PROPERTY_HEARTBEAT_INTERVAL_MS);

bsls::TimeInterval interval;
interval.addMilliseconds(heartbeatIntervalMs);
Expand Down
19 changes: 11 additions & 8 deletions src/groups/bmq/bmqimp/bmqimp_application.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,7 @@

#include <bmqimp_brokersession.h>
#include <bmqimp_eventqueue.h>
#include <bmqimp_negotiatedchannelfactory.h>
#include <bmqimp_initialconnectionchannelfactory.h>
#include <bmqp_ctrlmsg_messages.h>
#include <bmqp_heartbeatmonitor.h>
#include <bmqt_sessionoptions.h>
Expand Down Expand Up @@ -81,7 +81,7 @@ namespace bmqimp {
class Application {
public:
// PUBLIC TYPES
typedef bmqp::BlobPoolUtil::BlobSpPool BlobSpPool;
typedef bmqp::BlobPoolUtil::BlobSpPool BlobSpPool;
typedef bmqp::BlobPoolUtil::BlobSpPoolSp BlobSpPoolSp;

private:
Expand Down Expand Up @@ -134,7 +134,7 @@ class Application {

bmqio::StatChannelFactory d_statChannelFactory;

NegotiatedChannelFactory d_negotiatedChannelFactory;
InitialConnectionChannelFactory d_initialConnectionChannelFactory;

ChannelFactoryOpHandleMp d_connectHandle_mp;

Expand Down Expand Up @@ -226,7 +226,8 @@ class Application {
const bsl::shared_ptr<bmqp::HeartbeatMonitor>& monitor);
bsl::shared_ptr<bmqp::HeartbeatMonitor>
createMonitor(const bsl::shared_ptr<bmqio::Channel>& channel);
void startHeartbeat(const bsl::shared_ptr<bmqio::Channel>& channel,
void
startHeartbeat(const bsl::shared_ptr<bmqio::Channel>& channel,
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is a formatter bug @dorjesinpo encountered

const bsl::shared_ptr<bmqp::HeartbeatMonitor>& monitor);
void stopHeartbeat();

Expand All @@ -249,10 +250,12 @@ class Application {
/// negotiation. The application will use the specified
/// `eventHandlerCB`. Use the specified `allocator` for all memory
/// allocations.
Application(const bmqt::SessionOptions& sessionOptions,
const bmqp_ctrlmsg::NegotiationMessage& negotiationMessage,
const EventQueue::EventHandlerCallback& eventHandlerCB,
bslma::Allocator* allocator);
Application(
const bmqt::SessionOptions& sessionOptions,
const bmqp_ctrlmsg::AuthenticationMessage& authenticationMessage,
const bmqp_ctrlmsg::NegotiationMessage& negotiationMessage,
const EventQueue::EventHandlerCallback& eventHandlerCB,
bslma::Allocator* allocator);

/// Destructor
~Application();
Expand Down
9 changes: 9 additions & 0 deletions src/groups/bmq/bmqimp/bmqimp_application.t.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,11 +45,14 @@ static void test1_breathingTest()

// Create a default application
bmqt::SessionOptions options(bmqtst::TestHelperUtil::allocator());
bmqp_ctrlmsg::AuthenticationMessage authenticationMessage(
bmqtst::TestHelperUtil::allocator());
bmqp_ctrlmsg::NegotiationMessage negotiationMessage(
bmqtst::TestHelperUtil::allocator());
bmqimp::EventQueue::EventHandlerCallback emptyEventHandler;

bmqimp::Application obj(options,
authenticationMessage,
negotiationMessage,
emptyEventHandler,
bmqtst::TestHelperUtil::allocator());
Expand Down Expand Up @@ -78,11 +81,14 @@ static void test2_startStopTest()

// Create a default application, make sure it can start/stop
bmqt::SessionOptions options(bmqtst::TestHelperUtil::allocator());
bmqp_ctrlmsg::AuthenticationMessage authenticationMessage(
bmqtst::TestHelperUtil::allocator());
bmqp_ctrlmsg::NegotiationMessage negotiationMessage(
bmqtst::TestHelperUtil::allocator());
bmqimp::EventQueue::EventHandlerCallback emptyEventHandler;

bmqimp::Application obj(options,
authenticationMessage,
negotiationMessage,
emptyEventHandler,
bmqtst::TestHelperUtil::allocator());
Expand Down Expand Up @@ -130,11 +136,14 @@ static void test3_startStopAsyncTest()

// Create a default application, make sure it can start/stop
bmqt::SessionOptions options(bmqtst::TestHelperUtil::allocator());
bmqp_ctrlmsg::AuthenticationMessage authenticationMessage(
bmqtst::TestHelperUtil::allocator());
bmqp_ctrlmsg::NegotiationMessage negotiationMessage(
bmqtst::TestHelperUtil::allocator());
bmqimp::EventQueue::EventHandlerCallback emptyEventHandler;

bmqimp::Application obj(options,
authenticationMessage,
negotiationMessage,
emptyEventHandler,
bmqtst::TestHelperUtil::allocator());
Expand Down
8 changes: 4 additions & 4 deletions src/groups/bmq/bmqimp/bmqimp_brokersession.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@

#include <bmqscm_version.h>
// BMQ
#include <bmqimp_negotiatedchannelfactory.h>
#include <bmqimp_initialconnectionchannelfactory.h>
#include <bmqimp_queue.h>
#include <bmqp_ackeventbuilder.h>
#include <bmqp_ackmessageiterator.h>
Expand Down Expand Up @@ -409,7 +409,7 @@ void BrokerSession::SessionFsm::setStarted(
// Temporary safety switch to control configure request.
d_session.d_channel_sp->properties().load(
&d_session.d_doConfigureStream,
NegotiatedChannelFactory::k_CHANNEL_PROPERTY_CONFIGURE_STREAM);
InitialConnectionChannelFactory::k_CHANNEL_PROPERTY_CONFIGURE_STREAM);
}

bmqt::GenericResult::Enum
Expand Down Expand Up @@ -3621,7 +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
Expand Down Expand Up @@ -6043,7 +6043,7 @@ void BrokerSession::onOpenQueueResponse(

if (d_channel_sp->properties().load(
&isMPsEx,
NegotiatedChannelFactory::k_CHANNEL_PROPERTY_MPS_EX)) {
InitialConnectionChannelFactory::k_CHANNEL_PROPERTY_MPS_EX)) {
BSLS_ASSERT_SAFE(isMPsEx);
queue->setOldStyle(false);
}
Expand Down
Loading
Loading