You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by mm...@apache.org on 2023/01/19 21:56:47 UTC
[pulsar-client-cpp] branch branch-3.1 updated (9a4b0d1 -> 9350d1b)
This is an automated email from the ASF dual-hosted git repository.
mmerli pushed a change to branch branch-3.1
in repository https://gitbox.apache.org/repos/asf/pulsar-client-cpp.git
from 9a4b0d1 [fix] Fix a crash when closing a connection while connecting (#136)
new 5bded8b [fix] Use authoritative argument correctly in BinaryProtoLookupService::findBroker (#146)
new e6a8a64 Fix segfault caused by socket I/O on a closed io_service (#168)
new 9350d1b [fix] Fix hostname verification (#126)
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../setup-test-service-container.sh | 26 +++---
... => start-mim-test-service-inside-container.sh} | 49 ++---------
lib/BinaryProtoLookupService.cc | 4 +-
lib/ClientConnection.cc | 12 +--
lib/ClientConnection.h | 9 +-
pulsar-test-service-start.sh | 10 +--
test-conf/{client-ssl.conf => client-ssl-mim.conf} | 6 +-
test-conf/hn-verification/broker-cert.pem | 27 ++++++
test-conf/hn-verification/broker-key.pem | 28 +++++++
test-conf/hn-verification/cacert.pem | 29 +++++++
...standalone-ssl.conf => standalone-ssl-mim.conf} | 12 +--
tests/AuthPluginTest.cc | 97 +++++++++++++++++++++-
12 files changed, 224 insertions(+), 85 deletions(-)
copy pkg/apk/build-apk.sh => build-support/setup-test-service-container.sh (65%)
copy build-support/{start-test-service-inside-container.sh => start-mim-test-service-inside-container.sh} (51%)
copy test-conf/{client-ssl.conf => client-ssl-mim.conf} (87%)
create mode 100644 test-conf/hn-verification/broker-cert.pem
create mode 100644 test-conf/hn-verification/broker-key.pem
create mode 100644 test-conf/hn-verification/cacert.pem
copy test-conf/{standalone-ssl.conf => standalone-ssl-mim.conf} (98%)
[pulsar-client-cpp] 03/03: [fix] Fix hostname verification (#126)
Posted by mm...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mmerli pushed a commit to branch branch-3.1
in repository https://gitbox.apache.org/repos/asf/pulsar-client-cpp.git
commit 9350d1bfc80c56fe563e2d0211dcb8ec8aaf81e7
Author: Hideaki Oguni <22...@users.noreply.github.com>
AuthorDate: Tue Jan 17 22:09:21 2023 +0900
[fix] Fix hostname verification (#126)
### Motivation
If `ValidateHostName` is set to true, handshake always fails.
```
INFO [] ClientConnection:375 | [ -> ] Connected to broker
ERROR [] ClientConnection:463 | [ -> ] Handshake failed: certificate verify failed
INFO [] ClientConnection:1560 | [ -> ] Connection closed
```
### Modifications
- Verify `serviceUrl.host()`, not `physicalAddress`.
- `physicalAddress` is serviceUrl, which contains protocol (e.g. pulsar+ssl) and port number.
- Use `ssl::stream::set_verify_callback` instead of `ssl::context::set_verify_callback`.
- Verification should work with `ssl::context::set_verify_callback`, but somehow it doesn't work.
Co-authored-by: hoguni <ho...@yahoo-corp.jp>
---
.../setup-test-service-container.sh | 22 +-
.../start-mim-test-service-inside-container.sh | 66 +++++
lib/ClientConnection.cc | 10 +-
pulsar-test-service-start.sh | 10 +-
.../client-ssl-mim.conf | 26 +-
test-conf/hn-verification/broker-cert.pem | 27 ++
test-conf/hn-verification/broker-key.pem | 28 ++
test-conf/hn-verification/cacert.pem | 29 ++
test-conf/standalone-ssl-mim.conf | 309 +++++++++++++++++++++
tests/AuthPluginTest.cc | 97 ++++++-
10 files changed, 579 insertions(+), 45 deletions(-)
diff --git a/pulsar-test-service-start.sh b/build-support/setup-test-service-container.sh
similarity index 60%
copy from pulsar-test-service-start.sh
copy to build-support/setup-test-service-container.sh
index 1b439ec..a02a6d4 100755
--- a/pulsar-test-service-start.sh
+++ b/build-support/setup-test-service-container.sh
@@ -18,21 +18,19 @@
# under the License.
#
-set -e
+set -e -x
-SRC_DIR=$(git rev-parse --show-toplevel)
-cd $SRC_DIR
+if [ $# -ne 2 ]; then
+ echo "Usage: $0 \$CONTAINER_ID \$START_TEST_SERVICE_INSIDE_CONTAINER"
+ exit 1
+fi
-./pulsar-test-service-stop.sh
+CONTAINER_ID=$1
+START_TEST_SERVICE_INSIDE_CONTAINER=$2
-CONTAINER_ID=$(docker run -i -p 8080:8080 -p 6650:6650 -p 8443:8443 -p 6651:6651 --rm --detach apachepulsar/pulsar:latest sleep 3600)
-echo $CONTAINER_ID > .tests-container-id.txt
+echo $CONTAINER_ID >> .tests-container-id.txt
docker cp test-conf $CONTAINER_ID:/pulsar/test-conf
-docker cp build-support/start-test-service-inside-container.sh $CONTAINER_ID:start-test-service-inside-container.sh
+docker cp build-support/$START_TEST_SERVICE_INSIDE_CONTAINER $CONTAINER_ID:$START_TEST_SERVICE_INSIDE_CONTAINER
-docker exec -i $CONTAINER_ID /start-test-service-inside-container.sh
-
-docker cp $CONTAINER_ID:/pulsar/data/tokens/token.txt .test-token.txt
-
-echo "-- Ready to start tests"
+docker exec -i $CONTAINER_ID /$START_TEST_SERVICE_INSIDE_CONTAINER
diff --git a/build-support/start-mim-test-service-inside-container.sh b/build-support/start-mim-test-service-inside-container.sh
new file mode 100755
index 0000000..d265191
--- /dev/null
+++ b/build-support/start-mim-test-service-inside-container.sh
@@ -0,0 +1,66 @@
+#!/usr/bin/env bash
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+set -e -x
+
+export PULSAR_EXTRA_OPTS=-Dpulsar.auth.basic.conf=test-conf/.htpasswd
+
+# Generate secret key and token
+mkdir -p data/tokens
+bin/pulsar tokens create-secret-key --output data/tokens/secret.key
+
+bin/pulsar tokens create \
+ --subject token-principal \
+ --secret-key file:///pulsar/data/tokens/secret.key \
+ > /pulsar/data/tokens/token.txt
+
+export PULSAR_STANDALONE_CONF=test-conf/standalone-ssl-mim.conf
+export PULSAR_PID_DIR=/tmp
+bin/pulsar-daemon start standalone \
+ --no-functions-worker --no-stream-storage \
+ --bookkeeper-dir data/bookkeeper
+
+echo "-- Wait for Pulsar service to be ready"
+until curl http://localhost:8081/metrics > /dev/null 2>&1 ; do sleep 1; done
+
+echo "-- Pulsar service is ready -- Configure permissions"
+
+export PULSAR_CLIENT_CONF=test-conf/client-ssl-mim.conf
+
+# Create "standalone" cluster if it does not exist
+bin/pulsar-admin clusters list | grep -q '^standalone$' ||
+ bin/pulsar-admin clusters create \
+ standalone \
+ --url http://localhost:8081/ \
+ --url-secure https://localhost:8444/ \
+ --broker-url pulsar://localhost:6652/ \
+ --broker-url-secure pulsar+ssl://localhost:6653/
+
+# Create "private" tenant
+bin/pulsar-admin tenants create private -r "" -c "standalone"
+
+# Create "private/auth" with required authentication
+bin/pulsar-admin namespaces create private/auth --clusters standalone
+
+bin/pulsar-admin namespaces grant-permission private/auth \
+ --actions produce,consume \
+ --role "token-principal"
+
+echo "-- Ready to start tests"
diff --git a/lib/ClientConnection.cc b/lib/ClientConnection.cc
index 51aaeac..414348a 100644
--- a/lib/ClientConnection.cc
+++ b/lib/ClientConnection.cc
@@ -201,11 +201,6 @@ ClientConnection::ClientConnection(const std::string& logicalAddress, const std:
} else {
ctx.set_verify_mode(boost::asio::ssl::context::verify_peer);
- if (clientConfiguration.isValidateHostName()) {
- LOG_DEBUG("Validating hostname for " << serviceUrl.host() << ":" << serviceUrl.port());
- ctx.set_verify_callback(boost::asio::ssl::rfc2818_verification(physicalAddress));
- }
-
std::string trustCertFilePath = clientConfiguration.getTlsTrustCertsFilePath();
if (!trustCertFilePath.empty()) {
if (file_exists(trustCertFilePath)) {
@@ -254,6 +249,11 @@ ClientConnection::ClientConnection(const std::string& logicalAddress, const std:
tlsSocket_ = ExecutorService::createTlsSocket(socket_, ctx);
+ if (!clientConfiguration.isTlsAllowInsecureConnection() && clientConfiguration.isValidateHostName()) {
+ LOG_DEBUG("Validating hostname for " << serviceUrl.host() << ":" << serviceUrl.port());
+ tlsSocket_->set_verify_callback(boost::asio::ssl::rfc2818_verification(serviceUrl.host()));
+ }
+
LOG_DEBUG("TLS SNI Host: " << serviceUrl.host());
if (!SSL_set_tlsext_host_name(tlsSocket_->native_handle(), serviceUrl.host().c_str())) {
boost::system::error_code ec{static_cast<int>(::ERR_get_error()),
diff --git a/pulsar-test-service-start.sh b/pulsar-test-service-start.sh
index 1b439ec..f2f6c83 100755
--- a/pulsar-test-service-start.sh
+++ b/pulsar-test-service-start.sh
@@ -26,13 +26,11 @@ cd $SRC_DIR
./pulsar-test-service-stop.sh
CONTAINER_ID=$(docker run -i -p 8080:8080 -p 6650:6650 -p 8443:8443 -p 6651:6651 --rm --detach apachepulsar/pulsar:latest sleep 3600)
-echo $CONTAINER_ID > .tests-container-id.txt
-
-docker cp test-conf $CONTAINER_ID:/pulsar/test-conf
-docker cp build-support/start-test-service-inside-container.sh $CONTAINER_ID:start-test-service-inside-container.sh
-
-docker exec -i $CONTAINER_ID /start-test-service-inside-container.sh
+build-support/setup-test-service-container.sh $CONTAINER_ID start-test-service-inside-container.sh
docker cp $CONTAINER_ID:/pulsar/data/tokens/token.txt .test-token.txt
+CONTAINER_ID=$(docker run -i -p 8081:8081 -p 6652:6652 -p 8444:8444 -p 6653:6653 --rm --detach apachepulsar/pulsar:latest sleep 3600)
+build-support/setup-test-service-container.sh $CONTAINER_ID start-mim-test-service-inside-container.sh
+
echo "-- Ready to start tests"
diff --git a/pulsar-test-service-start.sh b/test-conf/client-ssl-mim.conf
old mode 100755
new mode 100644
similarity index 55%
copy from pulsar-test-service-start.sh
copy to test-conf/client-ssl-mim.conf
index 1b439ec..b04cd7a
--- a/pulsar-test-service-start.sh
+++ b/test-conf/client-ssl-mim.conf
@@ -1,4 +1,3 @@
-#!/usr/bin/env bash
#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
@@ -18,21 +17,10 @@
# under the License.
#
-set -e
-
-SRC_DIR=$(git rev-parse --show-toplevel)
-cd $SRC_DIR
-
-./pulsar-test-service-stop.sh
-
-CONTAINER_ID=$(docker run -i -p 8080:8080 -p 6650:6650 -p 8443:8443 -p 6651:6651 --rm --detach apachepulsar/pulsar:latest sleep 3600)
-echo $CONTAINER_ID > .tests-container-id.txt
-
-docker cp test-conf $CONTAINER_ID:/pulsar/test-conf
-docker cp build-support/start-test-service-inside-container.sh $CONTAINER_ID:start-test-service-inside-container.sh
-
-docker exec -i $CONTAINER_ID /start-test-service-inside-container.sh
-
-docker cp $CONTAINER_ID:/pulsar/data/tokens/token.txt .test-token.txt
-
-echo "-- Ready to start tests"
+# Pulsar Client configuration
+webServiceUrl=https://localhost:8444/
+brokerServiceUrl=pulsar+ssl://localhost:6653/
+tlsAllowInsecureConnection=false
+tlsTrustCertsFilePath=test-conf/hn-verification/cacert.pem
+authPlugin=org.apache.pulsar.client.impl.auth.AuthenticationTls
+authParams=tlsCertFile:test-conf/client-cert.pem,tlsKeyFile:test-conf/client-key.pem
diff --git a/test-conf/hn-verification/broker-cert.pem b/test-conf/hn-verification/broker-cert.pem
new file mode 100644
index 0000000..b5c7a5d
--- /dev/null
+++ b/test-conf/hn-verification/broker-cert.pem
@@ -0,0 +1,27 @@
+-----BEGIN CERTIFICATE-----
+MIIEkDCCAnigAwIBAgICEAAwDQYJKoZIhvcNAQELBQAwETEPMA0GA1UEAwwGZm9v
+YmFyMCAXDTE4MDYyMjA4NTUzMloYDzIyOTIwNDA2MDg1NTMyWjAjMSEwHwYDVQQD
+DBhicm9rZXIucHVsc2FyLmFwYWNoZS5vcmcwggEiMA0GCSqGSIb3DQEBAQUAA4IB
+DwAwggEKAoIBAQDQouKhZah4hMCqmg4aS5RhQG/Y1gA+yP9DGF9mlw35tfhfWs63
+EvNjEK4L/ZWSEV45L/wc6YV14RmM6bJ0V/0vXo4xmISbqptND/2kRIspkLZQ5F0O
+OQXVicqZLOc6igZQhRg8ANDYdTJUTF65DqauX4OJt3YMhF2FSt7jQtlj06IQBa01
++ARO9OotMJtBY+vIU5bV6JydfgkhQH9rIDI7AMeY5j02gGkJJrelfm+WoOsUez+X
+aqTN3/tF8+MBcFB3G04s1qc2CJPJM3YGxvxEtHqTGI14t9J8p5O7X9JHpcY8X00s
+bxa4FGbKgfDobbkJ+GgblWCkAcLN95sKTqtHAgMBAAGjgd0wgdowCQYDVR0TBAIw
+ADARBglghkgBhvhCAQEEBAMCBkAwMwYJYIZIAYb4QgENBCYWJE9wZW5TU0wgR2Vu
+ZXJhdGVkIFNlcnZlciBDZXJ0aWZpY2F0ZTAdBgNVHQ4EFgQUaxFvJrkEGqk8azTA
+DyVyTyTbJAIwQQYDVR0jBDowOIAUVwvpyyPov0c+UHo/RX6hGEOdFSehFaQTMBEx
+DzANBgNVBAMMBmZvb2JhcoIJANfih0+geeIMMA4GA1UdDwEB/wQEAwIFoDATBgNV
+HSUEDDAKBggrBgEFBQcDATANBgkqhkiG9w0BAQsFAAOCAgEA35QDGclHzQtHs3yQ
+ZzNOSKisg5srTiIoQgRzfHrXfkthNFCnBzhKjBxqk3EIasVtvyGuk0ThneC1ai3y
+ZK3BivnMZfm1SfyvieFoqWetsxohWfcpOSVkpvO37P6v/NmmaTIGkBN3gxKCx0QN
+zqApLQyNTM++X3wxetYH/afAGUrRmBGWZuJheQpB9yZ+FB6BRp8YuYIYBzANJyW9
+spvXW03TpqX2AIoRBoGMLzK72vbhAbLWiCIfEYREhbZVRkP+yvD338cWrILlOEur
+x/n8L/FTmbf7mXzHg4xaQ3zg/5+0OCPMDPUBE4xWDBAbZ82hgOcTqfVjwoPgo2V0
+fbbx6redq44J3Vn5d9Xhi59fkpqEjHpX4xebr5iMikZsNTJMeLh0h3uf7DstuO9d
+mfnF5j+yDXCKb9XzCsTSvGCN+spmUh6RfSrbkw8/LrRvBUpKVEM0GfKSnaFpOaSS
+efM4UEi72FRjszzHEkdvpiLhYvihINLJmDXszhc3fCi42be/DGmUhuhTZWynOPmp
+0N0V/8/sGT5gh4fGEtGzS/8xEvZwO9uDlccJiG8Pi+aO0/K9urB9nppd/xKWXv3C
+cib/QrW0Qow4TADWC1fnGYCpFzzaZ2esPL2MvzOYXnW4/AbEqmb6Weatluai64ZK
+3N2cGJWRyvpvvmbP2hKCa4eLgEc=
+-----END CERTIFICATE-----
diff --git a/test-conf/hn-verification/broker-key.pem b/test-conf/hn-verification/broker-key.pem
new file mode 100644
index 0000000..2b51d01
--- /dev/null
+++ b/test-conf/hn-verification/broker-key.pem
@@ -0,0 +1,28 @@
+-----BEGIN PRIVATE KEY-----
+MIIEvwIBADANBgkqhkiG9w0BAQEFAASCBKkwggSlAgEAAoIBAQDQouKhZah4hMCq
+mg4aS5RhQG/Y1gA+yP9DGF9mlw35tfhfWs63EvNjEK4L/ZWSEV45L/wc6YV14RmM
+6bJ0V/0vXo4xmISbqptND/2kRIspkLZQ5F0OOQXVicqZLOc6igZQhRg8ANDYdTJU
+TF65DqauX4OJt3YMhF2FSt7jQtlj06IQBa01+ARO9OotMJtBY+vIU5bV6Jydfgkh
+QH9rIDI7AMeY5j02gGkJJrelfm+WoOsUez+XaqTN3/tF8+MBcFB3G04s1qc2CJPJ
+M3YGxvxEtHqTGI14t9J8p5O7X9JHpcY8X00sbxa4FGbKgfDobbkJ+GgblWCkAcLN
+95sKTqtHAgMBAAECggEBALE1eMtfnk3nbAI74bih84D7C0Ug14p8jJv/qqBnsx4j
+WrgbWDMVrJa7Rym2FQHBMMfgIwKnso0iSeJvaPz683j1lk833YKe0VQOPgD1m0IN
+wV1J6mQ3OOZcKDIcerY1IBHqSmBEzR7dxIbnaxlCAX9gb0hdBK6zCwA5TMG5OQ5Y
+3cGOmevK5i2PiejhpruA8h7E48P1ATaGHUZif9YD724oi6AcilQ8H/DlOjZTvlmK
+r4aJ30f72NwGM8Ecet5CE2wyflAGtY0k+nChYkPRfy54u64Z/T9B53AvneFaj8jv
+yFepZgRTs2cWhEl0KQGuBHQ4+IeOfMt2LebhvjWW8YkCgYEA7BXVsnqPHKRDd8wP
+eNkolY4Fjdq4wu9ad+DaFiZcJuv7ugr+Kplltq6e4aU36zEdBYdPp/6KM/HGE/Xj
+bo0CELNUKs/Ny9H/UJc8DDbVEmoF3XGiIbKKq1T8NTXTETFnwrGkBFD8nl7YTsOF
+M4FZmSok0MhhkpEULAqxBS6YpQsCgYEA4jxM1egTVSWjTreg2UdYo2507jKa7maP
+PRtoPsNJzWNbOpfj26l3/8pd6oYKWck6se6RxIUxUrk3ywhNJIIOvWEC7TaOH1c9
+T4NQNcweqBW9+A1x5gyzT14gDaBfl45gs82vI+kcpVv/w2N3HZOQZX3yAUqWpfw2
+yw1uQDXtgDUCgYEAiYPWbBXTkp1j5z3nrT7g0uxc89n5USLWkYlZvxktCEbg4+dP
+UUT06EoipdD1F3wOKZA9p98uZT9pX2sUxOpBz7SFTEKq3xQ9IZZWFc9CoW08aVat
+V++FsnLYTa5CeXtLsy6CGTmLTDx2xrpAtlWb+QmBVFPD8fmrxFOd9STFKS0CgYAt
+6ztVN3OlFqyc75yQPXD6SxMkvdTAisSMDKIOCylRrNb5f5baIP2gR3zkeyxiqPtm
+3htsHfSy67EtXpP50wQW4Dft2eLi7ZweJXMEWFfomfEjBeeWYAGNHHe5DFIauuVZ
+2WexDEGqNpAlIm0s7aSjVPrn1DHbouOkNyenlMqN+QKBgQDVYVhk9widShSnCmUA
+G30moXDgj3eRqCf5T7NEr9GXD1QBD/rQSPh5agnDV7IYLpV7/wkYLI7l9x7mDwu+
+I9mRXkyAmTVEctLTdXQHt0jdJa5SfUaVEDUzQbr0fUjkmythTvqZ809+d3ELPeLI
+5qJ7jxgksHWji4lYfL4r4J6Zaw==
+-----END PRIVATE KEY-----
diff --git a/test-conf/hn-verification/cacert.pem b/test-conf/hn-verification/cacert.pem
new file mode 100644
index 0000000..0446700
--- /dev/null
+++ b/test-conf/hn-verification/cacert.pem
@@ -0,0 +1,29 @@
+-----BEGIN CERTIFICATE-----
+MIIFCDCCAvCgAwIBAgIJANfih0+geeIMMA0GCSqGSIb3DQEBCwUAMBExDzANBgNV
+BAMMBmZvb2JhcjAeFw0xODA2MjIwODQ2MjFaFw0zODA2MTcwODQ2MjFaMBExDzAN
+BgNVBAMMBmZvb2JhcjCCAiIwDQYJKoZIhvcNAQEBBQADggIPADCCAgoCggIBAOVU
+UpTPeXCeyfUiQS824l9s9krZd4R6TA4D97eQ9EWm2D7ppV4gPApHO8j5f+joo/b6
+Iso4aFlHpJ8VV2a5Ol7rjQw43MJHaBgwDxB1XWgsNdfoI7ebtp/BWg2nM3r8wm+Z
+gKenf9d1/1Ol+6yFUehkLkIXUvldiVegmmje8FnwhcDNE1eTrh66XqSJXEXqgBKu
+NqsoYcVak72OyOO1/N8CESoSdyBkbSiH5vJyo0AUCjn7tULga7fxojmqBZDog9Pg
+e5Fi/hbCrdinbxBrMgIxQ7wqXw2sw6iOWu4FU8Ih/CuF4xaQy2YP7MEk4Ff0LCY0
+KMhFMWU7550r/fz/C2l7fKhREyCQPa/bVE+dfxgZ/gCZ+p7vQ154hCCjpd+5bECv
+SN1bcVIPG6ngQu4vMXa7QRBi/Od40jSVGVJXYY6kXvrYatad7035w2GGGGkvMsQm
+y53yh4tqQfH7ulHqB0J5LebTQRp6nRizWigVCLjNkxJYI+Dj51qvT1zdyWEegKr1
+CthBfYzXlfjeH3xri1f0UABeC12n24Wkacd9af7zs7S3rYntEK444w/3fB0F62Lh
+SESfMLAmUH0dF5plRShrFUXz23nUeS8EYgWmnGkpf/HDzB67vdfAK0tfJEtmmY78
+q06OSgMr+AOOqaomh4Ez2ZQG592bS71G8MrE7r2/AgMBAAGjYzBhMB0GA1UdDgQW
+BBRXC+nLI+i/Rz5Qej9FfqEYQ50VJzAfBgNVHSMEGDAWgBRXC+nLI+i/Rz5Qej9F
+fqEYQ50VJzAPBgNVHRMBAf8EBTADAQH/MA4GA1UdDwEB/wQEAwIBhjANBgkqhkiG
+9w0BAQsFAAOCAgEAYd2PxdV+YOaWcmMG1fK7CGwSzDOGsgC7hi4gWPiNsVbz6fwQ
+m5Ac7Zw76dzin8gzOPKST7B8WIoc7ZWrMnyh3G6A3u29Ec8iWahqGa91NPA3bOIl
+0ldXnXfa416+JL/Q5utpiV6W2XDaB53v9GqpMk4rOTS9kCFOiuH5ZU8P69jp9mq6
+7pI/+hWFr+21ibmXH6ANxRLd/5+AqojRUYowAu2997Z+xmbpwx/2Svciq3LNY/Vz
+s9DudUHCBHj/DPgNxsEUt8QNohjQkRbFTY0a1aXodJ/pm0Ehk2kf9KwYYYduR7ak
+6UmPIPrZg6FePNahxwMZ0RtgX7EXmpiiIH1q9BsulddWkrFQclevsWO3ONQVrDs2
+gwY0HQuCRCJ+xgS2cyGiGohW5MkIsg1aI0i0j5GIUSppCIYgirAGCairARbCjhcx
+pbMe8RTuBhCqO3R2wZ0wXu7P7/ArI/Ltm1dU6IeHUAUmeneVj5ie0SdA19mHTS2o
+lG77N0jy6eq2zyEwJE6tuS/tyP1xrxdzXCYY7f6X9aNfsuPVQTcnrFajvDv8R6uD
+YnRStVCdS6fZEP0JzsLrqp9bgLIRRsiqsVVBCgJdK1I/X59qk2EyCLXWSgk8T9XZ
+iux8LlPpskt30YYt1KhlWB9zVz7k0uYAwits5foU6RfCRDPAyOa1q/QOXk0=
+-----END CERTIFICATE-----
diff --git a/test-conf/standalone-ssl-mim.conf b/test-conf/standalone-ssl-mim.conf
new file mode 100644
index 0000000..359645e
--- /dev/null
+++ b/test-conf/standalone-ssl-mim.conf
@@ -0,0 +1,309 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+### --- General broker settings --- ###
+
+# Zookeeper quorum connection string
+zookeeperServers=
+
+# Configuration Store connection string
+configurationStoreServers=
+
+brokerServicePort=6652
+brokerServicePortTls=6653
+
+# Port to use to server HTTP request
+webServicePort=8081
+webServicePortTls=8444
+
+# Hostname or IP address the service binds on, default is 0.0.0.0.
+bindAddress=0.0.0.0
+
+# Hostname or IP address the service advertises to the outside world. If not set, the value of InetAddress.getLocalHost().getCanonicalHostName() is used.
+advertisedAddress=localhost
+
+# Name of the cluster to which this broker belongs to
+clusterName=standalone
+
+# Zookeeper session timeout in milliseconds
+zooKeeperSessionTimeoutMillis=30000
+
+# Time to wait for broker graceful shutdown. After this time elapses, the process will be killed
+brokerShutdownTimeoutMs=3000
+
+# Enable backlog quota check. Enforces action on topic when the quota is reached
+backlogQuotaCheckEnabled=true
+
+# How often to check for topics that have reached the quota
+backlogQuotaCheckIntervalInSeconds=60
+
+# Default per-topic backlog quota limit
+backlogQuotaDefaultLimitGB=10
+
+# Enable the deletion of inactive topics
+brokerDeleteInactiveTopicsEnabled=true
+
+# How often to check for inactive topics
+brokerDeleteInactiveTopicsFrequencySeconds=60
+
+# How frequently to proactively check and purge expired messages
+messageExpiryCheckIntervalInMinutes=5
+
+# Enable check for minimum allowed client library version
+clientLibraryVersionCheckEnabled=false
+
+# Allow client libraries with no version information
+clientLibraryVersionCheckAllowUnversioned=true
+
+# Path for the file used to determine the rotation status for the broker when responding
+# to service discovery health checks
+statusFilePath=/usr/local/apache/htdocs
+
+# Max number of unacknowledged messages allowed to receive messages by a consumer on a shared subscription. Broker will stop sending
+# messages to consumer once, this limit reaches until consumer starts acknowledging messages back
+# Using a value of 0, is disabling unackeMessage limit check and consumer can receive messages without any restriction
+maxUnackedMessagesPerConsumer=50000
+
+subscriptionRedeliveryTrackerEnabled=true
+
+### --- Authentication --- ###
+
+# Enable TLS
+tlsEnabled=true
+tlsCertificateFilePath=test-conf/hn-verification/broker-cert.pem
+tlsKeyFilePath=test-conf/hn-verification/broker-key.pem
+tlsTrustCertsFilePath=test-conf/cacert.pem
+tlsAllowInsecureConnection=false
+
+anonymousUserRole=anonymous
+
+# Enable authentication
+authenticationEnabled=true
+
+# Authentication provider name list, which is comma separated list of class names
+authenticationProviders=org.apache.pulsar.broker.authentication.AuthenticationProviderTls,org.apache.pulsar.broker.authentication.AuthenticationProviderToken,org.apache.pulsar.broker.authentication.AuthenticationProviderBasic
+
+# Enforce authorization
+authorizationEnabled=true
+
+tokenSecretKey=file:///pulsar/data/tokens/secret.key
+
+# Role names that are treated as "super-user", meaning they will be able to do all admin
+# operations and publish/consume from all topics
+superUserRoles=localhost,superUser,admin
+
+# Authentication settings of the broker itself. Used when the broker connects to other brokers,
+# either in same or other clusters
+brokerClientAuthenticationPlugin=
+brokerClientAuthenticationParameters=
+
+### --- BookKeeper Client --- ###
+
+# Authentication plugin to use when connecting to bookies
+bookkeeperClientAuthenticationPlugin=
+
+# BookKeeper auth plugin implementatation specifics parameters name and values
+bookkeeperClientAuthenticationParametersName=
+bookkeeperClientAuthenticationParameters=
+
+# Timeout for BK add / read operations
+bookkeeperClientTimeoutInSeconds=30
+
+# Speculative reads are initiated if a read request doesn't complete within a certain time
+# Using a value of 0, is disabling the speculative reads
+bookkeeperClientSpeculativeReadTimeoutInMillis=0
+
+# Enable bookies health check. Bookies that have more than the configured number of failure within
+# the interval will be quarantined for some time. During this period, new ledgers won't be created
+# on these bookies
+bookkeeperClientHealthCheckEnabled=true
+bookkeeperClientHealthCheckIntervalSeconds=60
+bookkeeperClientHealthCheckErrorThresholdPerInterval=5
+bookkeeperClientHealthCheckQuarantineTimeInSeconds=1800
+
+# Enable rack-aware bookie selection policy. BK will chose bookies from different racks when
+# forming a new bookie ensemble
+bookkeeperClientRackawarePolicyEnabled=true
+
+# Enable region-aware bookie selection policy. BK will chose bookies from
+# different regions and racks when forming a new bookie ensemble
+# If enabled, the value of bookkeeperClientRackawarePolicyEnabled is ignored
+bookkeeperClientRegionawarePolicyEnabled=false
+
+# Minimum number of racks per write quorum. BK rack-aware bookie selection policy will try to
+# get bookies from at least 'bookkeeperClientMinNumRacksPerWriteQuorum' racks for a write quorum.
+bookkeeperClientMinNumRacksPerWriteQuorum=1
+
+# Enforces rack-aware bookie selection policy to pick bookies from 'bookkeeperClientMinNumRacksPerWriteQuorum'
+# racks for a writeQuorum.
+# If BK can't find bookie then it would throw BKNotEnoughBookiesException instead of picking random one.
+bookkeeperClientEnforceMinNumRacksPerWriteQuorum=false
+
+# Enable/disable reordering read sequence on reading entries.
+bookkeeperClientReorderReadSequenceEnabled=false
+
+# Enable bookie isolation by specifying a list of bookie groups to choose from. Any bookie
+# outside the specified groups will not be used by the broker
+bookkeeperClientIsolationGroups=
+
+### --- Managed Ledger --- ###
+
+# Number of bookies to use when creating a ledger
+managedLedgerDefaultEnsembleSize=1
+
+# Number of copies to store for each message
+managedLedgerDefaultWriteQuorum=1
+
+# Number of guaranteed copies (acks to wait before write is complete)
+managedLedgerDefaultAckQuorum=1
+
+# Amount of memory to use for caching data payload in managed ledger. This memory
+# is allocated from JVM direct memory and it's shared across all the topics
+# running in the same broker
+managedLedgerCacheSizeMB=1024
+
+# Threshold to which bring down the cache level when eviction is triggered
+managedLedgerCacheEvictionWatermark=0.9
+
+# Rate limit the amount of writes generated by consumer acking the messages
+managedLedgerDefaultMarkDeleteRateLimit=0.1
+
+# Max number of entries to append to a ledger before triggering a rollover
+# A ledger rollover is triggered after the min rollover time has passed
+# and one of the following conditions is true:
+# * The max rollover time has been reached
+# * The max entries have been written to the ledger
+# * The max ledger size has been written to the ledger
+managedLedgerMaxEntriesPerLedger=50000
+
+# Minimum time between ledger rollover for a topic
+managedLedgerMinLedgerRolloverTimeMinutes=10
+
+# Maximum time before forcing a ledger rollover for a topic
+managedLedgerMaxLedgerRolloverTimeMinutes=240
+
+# Max number of entries to append to a cursor ledger
+managedLedgerCursorMaxEntriesPerLedger=50000
+
+# Max time before triggering a rollover on a cursor ledger
+managedLedgerCursorRolloverTimeInSeconds=14400
+
+
+
+### --- Load balancer --- ###
+
+# Enable load balancer
+loadBalancerEnabled=false
+
+# Strategy to assign a new bundle
+loadBalancerPlacementStrategy=weightedRandomSelection
+
+# Percentage of change to trigger load report update
+loadBalancerReportUpdateThresholdPercentage=10
+
+# maximum interval to update load report
+loadBalancerReportUpdateMaxIntervalMinutes=15
+
+# Frequency of report to collect
+loadBalancerHostUsageCheckIntervalMinutes=1
+
+# Load shedding interval. Broker periodically checks whether some traffic should be offload from
+# some over-loaded broker to other under-loaded brokers
+loadBalancerSheddingIntervalMinutes=30
+
+# Prevent the same topics to be shed and moved to other broker more than once within this timeframe
+loadBalancerSheddingGracePeriodMinutes=30
+
+# Usage threshold to determine a broker as under-loaded
+loadBalancerBrokerUnderloadedThresholdPercentage=1
+
+# Usage threshold to determine a broker as over-loaded
+loadBalancerBrokerOverloadedThresholdPercentage=85
+
+# Interval to update namespace bundle resource quota
+loadBalancerResourceQuotaUpdateIntervalMinutes=15
+
+# Usage threshold to determine a broker is having just right level of load
+loadBalancerBrokerComfortLoadLevelPercentage=65
+
+# enable/disable namespace bundle auto split
+loadBalancerAutoBundleSplitEnabled=false
+
+# interval to detect & split hot namespace bundle
+loadBalancerNamespaceBundleSplitIntervalMinutes=15
+
+# maximum topics in a bundle, otherwise bundle split will be triggered
+loadBalancerNamespaceBundleMaxTopics=1000
+
+# maximum sessions (producers + consumers) in a bundle, otherwise bundle split will be triggered
+loadBalancerNamespaceBundleMaxSessions=1000
+
+# maximum msgRate (in + out) in a bundle, otherwise bundle split will be triggered
+loadBalancerNamespaceBundleMaxMsgRate=1000
+
+# maximum bandwidth (in + out) in a bundle, otherwise bundle split will be triggered
+loadBalancerNamespaceBundleMaxBandwidthMbytes=100
+
+# maximum number of bundles in a namespace
+loadBalancerNamespaceMaximumBundles=128
+
+### --- Replication --- ###
+
+# Enable replication metrics
+replicationMetricsEnabled=true
+
+# Max number of connections to open for each broker in a remote cluster
+# More connections host-to-host lead to better throughput over high-latency
+# links.
+replicationConnectionsPerBroker=16
+
+# Replicator producer queue size
+replicationProducerQueueSize=1000
+
+# Default message retention time. 0 means retention is disabled. -1 means data is not removed by time quota
+defaultRetentionTimeInMinutes=0
+
+# Default retention size. 0 means retention is disabled. -1 means data is not removed by size quota
+defaultRetentionSizeInMB=0
+
+# How often to check whether the connections are still alive
+keepAliveIntervalSeconds=30
+
+# Enable topic auto creation if new producer or consumer connected (disable auto creation with value false)
+allowAutoTopicCreation=true
+
+# The type of topic that is allowed to be automatically created.(partitioned/non-partitioned)
+allowAutoTopicCreationType=non-partitioned
+
+# The number of partitioned topics that is allowed to be automatically created if allowAutoTopicCreationType is partitioned.
+defaultNumPartitions=1
+
+### --- Deprecated config variables --- ###
+
+# Deprecated. Use configurationStoreServers
+globalZookeeperServers=
+
+# Deprecated. Use brokerDeleteInactiveTopicsFrequencySeconds
+brokerServicePurgeInactiveFrequencyInSeconds=60
+
+# Given a specific limit of the max message size
+maxMessageSize=1024000
+
+# Disable consistent hashing to fix flaky `KeySharedConsumerTest#testMultiTopics`.
+subscriptionKeySharedUseConsistentHashing=false
diff --git a/tests/AuthPluginTest.cc b/tests/AuthPluginTest.cc
index e2d3351..2d23bf9 100644
--- a/tests/AuthPluginTest.cc
+++ b/tests/AuthPluginTest.cc
@@ -41,6 +41,12 @@ static const std::string caPath = "../test-conf/cacert.pem";
static const std::string clientPublicKeyPath = "../test-conf/client-cert.pem";
static const std::string clientPrivateKeyPath = "../test-conf/client-key.pem";
+// Man in middle certificate which tries to act as a broker by sending its own valid certificate
+static const std::string mimServiceUrlTls = "pulsar+ssl://localhost:6653";
+static const std::string mimServiceUrlHttps = "https://localhost:8444";
+
+static const std::string mimCaPath = "../test-conf/hn-verification/cacert.pem";
+
static void sendCallBackTls(Result r, const MessageId& msgId) {
ASSERT_EQ(r, ResultOk);
globalTestTlsMessagesCounter++;
@@ -149,12 +155,54 @@ TEST(AuthPluginTest, testTlsDetectPulsarSslWithHostNameValidation) {
Producer producer;
Result res = client.createProducer(topicName, producer);
+ ASSERT_EQ(ResultOk, res);
+}
+
+TEST(AuthPluginTest, testTlsDetectPulsarSslWithHostNameValidationMissingCertsFile) {
+ ClientConfiguration config = ClientConfiguration();
+ config.setTlsAllowInsecureConnection(false);
+ config.setValidateHostName(true);
+ config.setAuth(pulsar::AuthTls::create(clientPublicKeyPath, clientPrivateKeyPath));
+
+ Client client(serviceUrlTls, config);
+ std::string topicName =
+ "persistent://private/auth/testTlsDetectPulsarSslWithHostNameValidationMissingCertsFile";
+
+ Producer producer;
+ Result res = client.createProducer(topicName, producer);
+ ASSERT_EQ(ResultConnectError, res);
+}
+
+TEST(AuthPluginTest, testTlsDetectPulsarSslWithInvalidBroker) {
+ ClientConfiguration configWithValidateHostname = ClientConfiguration();
+ configWithValidateHostname.setTlsTrustCertsFilePath(mimCaPath);
+ configWithValidateHostname.setTlsAllowInsecureConnection(false);
+ configWithValidateHostname.setValidateHostName(true);
+ configWithValidateHostname.setAuth(pulsar::AuthTls::create(clientPublicKeyPath, clientPrivateKeyPath));
+
+ ClientConfiguration config = ClientConfiguration();
+ config.setTlsTrustCertsFilePath(mimCaPath);
+ config.setTlsAllowInsecureConnection(false);
+ config.setAuth(pulsar::AuthTls::create(clientPublicKeyPath, clientPrivateKeyPath));
+
+ Client clientWithValidateHostname(mimServiceUrlTls, configWithValidateHostname);
+ Client client(mimServiceUrlTls, config);
+
+ std::string topicName = "persistent://private/auth/testTlsDetectPulsarSslWithInvalidBroker";
+
+ // 1. Client tries to connect to broker with hostname="localhost"
+ // 2. Broker sends x509 certificates with CN = "pulsar"
+ // 3. Client verifies the host-name and closes the connection
+ Producer producer;
+ Result res = clientWithValidateHostname.createProducer(topicName, producer);
ASSERT_EQ(ResultConnectError, res);
+
+ res = client.createProducer(topicName, producer);
+ ASSERT_EQ(ResultOk, res);
}
TEST(AuthPluginTest, testTlsDetectHttps) {
ClientConfiguration config = ClientConfiguration();
- config.setUseTls(true); // shouldn't be needed soon
config.setTlsTrustCertsFilePath(caPath);
config.setTlsAllowInsecureConnection(false);
config.setAuth(pulsar::AuthTls::create(clientPublicKeyPath, clientPrivateKeyPath));
@@ -173,7 +221,6 @@ TEST(AuthPluginTest, testTlsDetectHttps) {
TEST(AuthPluginTest, testTlsDetectHttpsWithHostNameValidation) {
ClientConfiguration config = ClientConfiguration();
- config.setUseTls(true); // shouldn't be needed soon
config.setTlsTrustCertsFilePath(caPath);
config.setTlsAllowInsecureConnection(false);
config.setAuth(pulsar::AuthTls::create(clientPublicKeyPath, clientPrivateKeyPath));
@@ -185,7 +232,51 @@ TEST(AuthPluginTest, testTlsDetectHttpsWithHostNameValidation) {
Producer producer;
Result res = client.createProducer(topicName, producer);
- ASSERT_NE(ResultOk, res);
+ ASSERT_EQ(ResultOk, res);
+}
+
+TEST(AuthPluginTest, testTlsDetectHttpsWithHostNameValidationMissingCertsFile) {
+ ClientConfiguration config = ClientConfiguration();
+ config.setTlsAllowInsecureConnection(false);
+ config.setAuth(pulsar::AuthTls::create(clientPublicKeyPath, clientPrivateKeyPath));
+ config.setValidateHostName(true);
+
+ Client client(serviceUrlHttps, config);
+
+ std::string topicName =
+ "persistent://private/auth/test-tls-detect-https-with-hostname-validation-missing-certs-file";
+
+ Producer producer;
+ Result res = client.createProducer(topicName, producer);
+ ASSERT_EQ(ResultLookupError, res);
+}
+
+TEST(AuthPluginTest, testTlsDetectHttpsWithInvalidBroker) {
+ ClientConfiguration configWithValidateHostname = ClientConfiguration();
+ configWithValidateHostname.setTlsTrustCertsFilePath(mimCaPath);
+ configWithValidateHostname.setTlsAllowInsecureConnection(false);
+ configWithValidateHostname.setValidateHostName(true);
+ configWithValidateHostname.setAuth(pulsar::AuthTls::create(clientPublicKeyPath, clientPrivateKeyPath));
+
+ ClientConfiguration config = ClientConfiguration();
+ config.setTlsTrustCertsFilePath(mimCaPath);
+ config.setTlsAllowInsecureConnection(false);
+ config.setAuth(pulsar::AuthTls::create(clientPublicKeyPath, clientPrivateKeyPath));
+
+ Client clientWithValidateHostname(mimServiceUrlHttps, configWithValidateHostname);
+ Client client(mimServiceUrlHttps, config);
+
+ std::string topicName = "persistent://private/auth/test-tls-detect-https-with-invalid-broker";
+
+ // 1. Client tries to connect to broker with hostname="localhost"
+ // 2. Broker sends x509 certificates with CN = "pulsar"
+ // 3. Client verifies the host-name and closes the connection
+ Producer producer;
+ Result res = clientWithValidateHostname.createProducer(topicName, producer);
+ ASSERT_EQ(ResultLookupError, res);
+
+ res = client.createProducer(topicName, producer);
+ ASSERT_EQ(ResultOk, res);
}
namespace testAthenz {
[pulsar-client-cpp] 01/03: [fix] Use authoritative argument correctly in BinaryProtoLookupService::findBroker (#146)
Posted by mm...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mmerli pushed a commit to branch branch-3.1
in repository https://gitbox.apache.org/repos/asf/pulsar-client-cpp.git
commit 5bded8b25d02b6207a70437f00f0580518acb5b0
Author: erobot <er...@users.noreply.github.com>
AuthorDate: Fri Dec 16 13:57:45 2022 +0800
[fix] Use authoritative argument correctly in BinaryProtoLookupService::findBroker (#146)
### Motivation
Use authoritative argument correctly in BinaryProtoLookupService::findBroker.
In the current code, authoritative field of lookup request is always false. When a bundle is not loaded by any broker, a lookup request with authoritative as false will not trigger broker to load the bundle, and lookup requests will loop between leader broker and the chosen broker. In this case, the lookup will not succeed and result in a large number of lookup requests to the broker as current c++ client do not have lookup redirect limit. If broker return failure after a large number [...]
of future will result in a large depth of function calls and may cause the stack to overflow.
https://github.com/apache/pulsar-client-cpp/blob/44f3b5240732a035395ec34b977301435dd0ec71/lib/BinaryProtoLookupService.cc#L55
### Modifications
Use authoritative argument.
---
lib/BinaryProtoLookupService.cc | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/lib/BinaryProtoLookupService.cc b/lib/BinaryProtoLookupService.cc
index b863d52..b925db4 100644
--- a/lib/BinaryProtoLookupService.cc
+++ b/lib/BinaryProtoLookupService.cc
@@ -38,7 +38,7 @@ auto BinaryProtoLookupService::findBroker(const std::string& address, bool autho
LOG_DEBUG("find broker from " << address << ", authoritative: " << authoritative << ", topic: " << topic);
auto promise = std::make_shared<Promise<Result, LookupResult>>();
// NOTE: we can use move capture for topic since C++14
- cnxPool_.getConnectionAsync(address).addListener([this, promise, topic, address](
+ cnxPool_.getConnectionAsync(address).addListener([this, promise, topic, address, authoritative](
Result result,
const ClientConnectionWeakPtr& weakCnx) {
if (result != ResultOk) {
@@ -52,7 +52,7 @@ auto BinaryProtoLookupService::findBroker(const std::string& address, bool autho
return;
}
auto lookupPromise = std::make_shared<LookupDataResultPromise>();
- cnx->newTopicLookup(topic, false, listenerName_, newRequestId(), lookupPromise);
+ cnx->newTopicLookup(topic, authoritative, listenerName_, newRequestId(), lookupPromise);
lookupPromise->getFuture().addListener([this, cnx, promise, topic, address](
Result result, const LookupDataResultPtr& data) {
if (result != ResultOk || !data) {
[pulsar-client-cpp] 02/03: Fix segfault caused by socket I/O on a closed io_service (#168)
Posted by mm...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mmerli pushed a commit to branch branch-3.1
in repository https://gitbox.apache.org/repos/asf/pulsar-client-cpp.git
commit e6a8a64880d99ddd1be5106336309ab06094b127
Author: Yunze Xu <xy...@163.com>
AuthorDate: Fri Jan 13 16:22:15 2023 +0800
Fix segfault caused by socket I/O on a closed io_service (#168)
Fixes https://github.com/apache/pulsar-client-cpp/issues/167
### Motivation
Here are some debugging info when the segfault happened in
`testCloseClient`. The outputs have been trimmed to make them clear.
An example crash at `async_write`:
```
#12 0x00007ffff7496dad in basic_stream_socket<...>::boost::asio::async_write /usr/include/boost/asio/impl/write.hpp:512
#13 0x00007ffff748e003 in ClientConnection::asyncWrite lib/ClientConnection.h:245
#14 0x00007ffff746e0b6 in ClientConnection::handleHandshake (this=0x555555e689d0) lib/ClientConnection.cc:502
```
Another example crash at `async_receive`:
```
#6 0x00007ffff7497247 in basic_stream_socket<...>::async_receive /usr/include/boost/asio/basic_stream_socket.hpp:677
#7 0x00007ffff748e647 in ClientConnection::asyncReceive lib/ClientConnection.h:258
#8 0x00007ffff746fa5d in ClientConnection::readNextCommand lib/ClientConnection.cc:606
```
The frame where it crashed:
```
245 if (descriptor_data->shutdown_)
(gdb) p descriptor_data
$2 = (boost::asio::detail::epoll_reactor::per_descriptor_data &) @0x555555e4a780: 0x0
```
We can see the socket descriptor is `nullptr`. The root cause is when
`async_receive` or `async_write` is called, the `io_service` object
might be closed. This case happened when `createProducerAsync` is
called, the actual producer creation continues in another thread, while
the `client.close()` happens in the current thread.
### Modifications
Check if the `ClientConnection` is closed before `async_receive` or
`async_write`. To avoid the use of lock, changing the `state_` field to
atomic.
### Verifications
```bash
./tests/pulsar-tests --gtest_filter='ClientTest.testCloseClient' --gtest_repeat=20
```
It never crashed after applying this patch.
---
lib/ClientConnection.cc | 2 +-
lib/ClientConnection.h | 9 ++++++++-
2 files changed, 9 insertions(+), 2 deletions(-)
diff --git a/lib/ClientConnection.cc b/lib/ClientConnection.cc
index 5c14467..51aaeac 100644
--- a/lib/ClientConnection.cc
+++ b/lib/ClientConnection.cc
@@ -811,7 +811,7 @@ void ClientConnection::handleIncomingMessage(const proto::CommandMessage& msg, b
void ClientConnection::handleIncomingCommand(BaseCommand& incomingCmd) {
LOG_DEBUG(cnxString_ << "Handling incoming command: " << Commands::messageType(incomingCmd.type()));
- switch (state_) {
+ switch (state_.load()) {
case Pending: {
LOG_ERROR(cnxString_ << "Connection is not ready yet");
break;
diff --git a/lib/ClientConnection.h b/lib/ClientConnection.h
index a07e2cd..91c5e88 100644
--- a/lib/ClientConnection.h
+++ b/lib/ClientConnection.h
@@ -22,6 +22,7 @@
#include <pulsar/ClientConfiguration.h>
#include <pulsar/defines.h>
+#include <atomic>
#include <boost/any.hpp>
#include <boost/asio/bind_executor.hpp>
#include <boost/asio/deadline_timer.hpp>
@@ -235,6 +236,9 @@ class PULSAR_PUBLIC ClientConnection : public std::enable_shared_from_this<Clien
template <typename ConstBufferSequence, typename WriteHandler>
inline void asyncWrite(const ConstBufferSequence& buffers, WriteHandler handler) {
+ if (isClosed()) {
+ return;
+ }
if (tlsSocket_) {
#if BOOST_VERSION >= 106600
boost::asio::async_write(*tlsSocket_, buffers, boost::asio::bind_executor(strand_, handler));
@@ -248,6 +252,9 @@ class PULSAR_PUBLIC ClientConnection : public std::enable_shared_from_this<Clien
template <typename MutableBufferSequence, typename ReadHandler>
inline void asyncReceive(const MutableBufferSequence& buffers, ReadHandler handler) {
+ if (isClosed()) {
+ return;
+ }
if (tlsSocket_) {
#if BOOST_VERSION >= 106600
tlsSocket_->async_read_some(buffers, boost::asio::bind_executor(strand_, handler));
@@ -259,7 +266,7 @@ class PULSAR_PUBLIC ClientConnection : public std::enable_shared_from_this<Clien
}
}
- State state_ = Pending;
+ std::atomic<State> state_{Pending};
TimeDuration operationsTimeout_;
AuthenticationPtr authentication_;
int serverProtocolVersion_;