Skip to content

Commit

Permalink
fixup! Support TLS listeners
Browse files Browse the repository at this point in the history
  • Loading branch information
@hallfox
hallfox committed Jan 10, 2025
1 parent d058805 commit 634c1e4
Show file tree
Hide file tree
Showing 12 changed files with 894 additions and 600 deletions.
4 changes: 2 additions & 2 deletions src/applications/bmqbrkr/etc/bmqbrkrcfg.json
View file
Original file line number Diff line number Diff line change
Expand Up @@ -8,8 +8,8 @@
"rotationBytes": 268435456,
"logfileFormat": "%d (%t) %s %F:%l %m\n\n",
"consoleFormat": "%d (%t) %s %F:%l %m\n",
"loggingVerbosity": "TRACE",
"consoleSeverityThreshold": "TRACE",
"loggingVerbosity": "INFO",
"consoleSeverityThreshold": "INFO",
"categories": [
"BMQBRKR:INFO:green",
"BMQ*:INFO:green",
Expand Down
229 changes: 134 additions & 95 deletions src/groups/bmq/bmqimp/bmqimp_application.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -135,6 +135,120 @@ ntcCreateInterfaceConfig(const bmqt::SessionOptions& sessionOptions,

} // close unnamed namespace

// -------------------------
// class bmqimp::Application::ChannelFactoryPipeline
// -------------------------

Application::ChannelFactoryPipeline::ChannelFactoryPipeline(
const bmqt::SessionOptions& sessionOptions,
bdlbb::BlobBufferFactory* blobBufferFactory,
bdlmt::EventScheduler* scheduler,
const bmqio::StatChannelFactoryConfig::StatContextCreatorFn&
statContextCreator,
const bmqp_ctrlmsg::NegotiationMessage& negotiationMessage,
BlobSpPool* blobSpPool,
bslma::Allocator* allocator)
: d_allocator_p(bslma::Default::allocator(allocator))
, d_channelFactory(ntcCreateInterfaceConfig(sessionOptions, d_allocator_p),
blobBufferFactory,
d_allocator_p)
, d_resolvingChannelFactory(
bmqio::ResolvingChannelFactoryConfig(
&d_channelFactory,
bmqex::ExecutionPolicyUtil::oneWay().alwaysBlocking().useExecutor(
bmqex::SystemExecutor())),
d_allocator_p)
, d_reconnectingChannelFactory(
bmqio::ReconnectingChannelFactoryConfig(&d_resolvingChannelFactory,
scheduler,
d_allocator_p),
d_allocator_p)
, d_statChannelFactory(
bmqio::StatChannelFactoryConfig(&d_reconnectingChannelFactory,
statContextCreator,
d_allocator_p),
d_allocator_p)
, d_negotiatedChannelFactory(
NegotiatedChannelFactoryConfig(&d_statChannelFactory,
negotiationMessage,
sessionOptions.connectTimeout(),
blobSpPool,
d_allocator_p),
d_allocator_p)
{
}

int Application::ChannelFactoryPipeline::configureTls(
const bsl::string& caPath)
{
ntca::EncryptionClientOptions encryptionClientOptions;

// TODO: Configure TLS protocol versions

// Set the minimum version to TLS 1.3
encryptionClientOptions.setMinMethod(ntca::EncryptionMethod::e_TLS_V1_3);
encryptionClientOptions.setMaxMethod(ntca::EncryptionMethod::e_DEFAULT);

// Enable server side authentication
encryptionClientOptions.setAuthentication(
ntca::EncryptionAuthentication::e_VERIFY);

encryptionClientOptions.addAuthorityFile(caPath);

ntsa::Error err = d_channelFactory.configureEncryptionClient(
encryptionClientOptions);

return err.code();
}

void Application::ChannelFactoryPipeline::listen(
bmqio::Status* status,
bslma::ManagedPtr<OpHandle>* handle,
const bmqio::ListenOptions& options,
const ResultCallback& cb)
{
d_negotiatedChannelFactory.listen(status, handle, options, cb);
}

void Application::ChannelFactoryPipeline::connect(
bmqio::Status* status,
bslma::ManagedPtr<OpHandle>* handle,
const bmqio::ConnectOptions& options,
const ResultCallback& cb)
{
d_negotiatedChannelFactory.connect(status, handle, options, cb);
}

int Application::ChannelFactoryPipeline::start()
{
// Start the channel factories.
int rc = d_channelFactory.start();
if (rc != 0) {
BALL_LOG_ERROR << "Failed to start channelFactory [rc: " << rc << "]";
return bmqt::GenericResult::e_UNKNOWN; // RETURN
}
bdlb::ScopeExitAny tcpScopeGuard(
bdlf::BindUtil::bind(&bmqio::NtcChannelFactory::stop,
&d_channelFactory));

rc = d_reconnectingChannelFactory.start();
if (rc != 0) {
BALL_LOG_ERROR << "Failed to start reconnectingChannelFactory [rc: "
<< rc << "]";
return bmqt::GenericResult::e_UNKNOWN; // RETURN
}

tcpScopeGuard.release();

return bmqt::GenericResult::e_SUCCESS;
}

void Application::ChannelFactoryPipeline::stop()
{
d_reconnectingChannelFactory.stop();
d_channelFactory.stop();
}

// -------------------------
// class bmqimp::Application
// -------------------------
Expand Down Expand Up @@ -303,8 +417,7 @@ void Application::brokerSessionStopped(
// This code assumes that there is no need to stop both factories upon
// e_START_FAILURE.
// If we wanted that, we would need another event.
d_reconnectingChannelFactory.stop();
d_channelFactory.stop();
d_channelFactoryPipeline.stop();
}

d_scheduler.cancelAllEventsAndWait();
Expand All @@ -320,71 +433,23 @@ void Application::brokerSessionStopped(
BALL_LOG_INFO << "bmqimp::Application stop completed";
}

int Application::loadTlsConfig(bmqio::NtcChannelFactory* channelFactory,
const bsl::string& caPath)
{
bmqio::NtcCertificateLoader certLoader =
channelFactory->createCertificateLoader();

int rc = 0;
if ((rc = d_certificateStore.loadCertificateAuthority(certLoader,
caPath))) {
return rc;
}

ntca::EncryptionClientOptions encryptionClientOptions;
// Set the minimum version to TLS 1.3
encryptionClientOptions.setMinMethod(ntca::EncryptionMethod::e_TLS_V1_3);
encryptionClientOptions.setMaxMethod(ntca::EncryptionMethod::e_TLS_V1_3);
// Enable server side authentication
encryptionClientOptions.setAuthentication(
ntca::EncryptionAuthentication::e_VERIFY);

ntsa::Error err;
{
bsl::vector<char> authorityData(&d_allocator);
err = d_certificateStore.certificateAuthority()->encode(
&authorityData);
if (err) {
return err.code();
}
encryptionClientOptions.addAuthorityData(authorityData);
}

err = channelFactory->createEncryptionClient(&d_encryptionClient_sp,
encryptionClientOptions);

return err.code();
}

bmqt::GenericResult::Enum Application::startChannel()
{
// executed by the FSM thread

BSLS_ASSERT_SAFE(d_brokerSession.state() ==
bmqimp::BrokerSession::State::e_STARTING);

int rc = 0;

// Start the channel factories.
rc = d_channelFactory.start();
int rc = d_channelFactoryPipeline.start();
if (rc != 0) {
BALL_LOG_ERROR << "Failed to start channelFactory [rc: " << rc << "]";
BALL_LOG_ERROR << "Failed to start channelFactoryPipeline [rc: " << rc
<< "]";
return bmqt::GenericResult::e_UNKNOWN; // RETURN
}
bdlb::ScopeExitAny tcpScopeGuard(
bdlf::BindUtil::bind(&bmqio::NtcChannelFactory::stop,
&d_channelFactory));

rc = d_reconnectingChannelFactory.start();
if (rc != 0) {
BALL_LOG_ERROR << "Failed to start reconnectingChannelFactory [rc: "
<< rc << "]";
return bmqt::GenericResult::e_UNKNOWN; // RETURN
}
bdlb::ScopeExitAny reconnectingScopeGuard(
bdlf::BindUtil::bind(&bmqio::ReconnectingChannelFactory::stop,
&d_reconnectingChannelFactory));
bdlb::ScopeExitAny pipelineScopeGuard(
bdlf::BindUtil::bind(&ChannelFactoryPipeline::stop,
&d_channelFactoryPipeline));

// Connect to the broker.
bmqio::TCPEndpoint endpoint(d_sessionOptions.brokerUri());
Expand All @@ -408,12 +473,7 @@ bmqt::GenericResult::Enum Application::startChannel()
.setAttemptInterval(attemptInterval)
.setAutoReconnect(true);

if (d_sessionOptions.isTlsSession()) {
loadTlsConfig(&d_channelFactory,
d_sessionOptions.certificateAuthority());
}

d_negotiatedChannelFactory.connect(
d_channelFactoryPipeline.connect(
&status,
&d_connectHandle_mp,
options,
Expand Down Expand Up @@ -447,8 +507,7 @@ bmqt::GenericResult::Enum Application::startChannel()
bdlf::MemFnUtil::memFn(&Application::snapshotStats, this));
}

reconnectingScopeGuard.release();
tcpScopeGuard.release();
pipelineScopeGuard.release();

return bmqt::GenericResult::e_SUCCESS;
}
Expand Down Expand Up @@ -603,37 +662,17 @@ Application::Application(
bmqp::BlobPoolUtil::createBlobPool(&d_blobBufferFactory,
d_allocators.get("BlobSpPool")))
, d_scheduler(bsls::SystemClockType::e_MONOTONIC, &d_allocator)
, d_channelFactory(ntcCreateInterfaceConfig(sessionOptions, allocator),
&d_blobBufferFactory,
allocator)
, d_resolvingChannelFactory(
bmqio::ResolvingChannelFactoryConfig(
&d_channelFactory,
bmqex::ExecutionPolicyUtil::oneWay().alwaysBlocking().useExecutor(
bmqex::SystemExecutor())),
allocator)
, d_reconnectingChannelFactory(
bmqio::ReconnectingChannelFactoryConfig(&d_resolvingChannelFactory,
&d_scheduler,
allocator),
allocator)
, d_statChannelFactory(
bmqio::StatChannelFactoryConfig(
&d_reconnectingChannelFactory,
bdlf::BindUtil::bind(&Application::channelStatContextCreator,
this,
bdlf::PlaceHolders::_1, // channel
bdlf::PlaceHolders::_2), // handle
allocator),
allocator)
, d_negotiatedChannelFactory(
NegotiatedChannelFactoryConfig(&d_statChannelFactory,
negotiationMessage,
sessionOptions.connectTimeout(),
d_blobSpPool_sp.get(),
d_encryptionClient_sp,
allocator),
allocator)
, d_channelFactoryPipeline(
sessionOptions,
&d_blobBufferFactory,
&d_scheduler,
bdlf::BindUtil::bind(&Application::channelStatContextCreator,
this,
bdlf::PlaceHolders::_1, // channel
bdlf::PlaceHolders::_2), // handle
negotiationMessage,
d_blobSpPool_sp.get(),
&d_allocator)
, d_connectHandle_mp()
, d_brokerSession(&d_scheduler,
&d_blobBufferFactory,
Expand Down
Loading

0 comments on commit 634c1e4

Please sign in to comment.