Merge pull request #3896 from bluca/fuzzers

Problems: memory sanitizer fails due to uninitialised global, want to run more tests with sanitizer

.travis.yml

@@ -6,7 +6,7 @@ os:
 - linux
 - osx
 
-dist: trusty
+dist: bionic
 
 cache: ccache
 
@@ -96,16 +96,12 @@ matrix:
         - libbsd-dev
   - env: BUILD_TYPE=default CURVE=libsodium DRAFT=enabled
     os: osx
-  - env: BUILD_TYPE=default CURVE=tweetnacl DRAFT=enabled ADDRESS_SANITIZER=enabled
+  - env: BUILD_TYPE=default CURVE=libsodium DRAFT=enabled ADDRESS_SANITIZER=enabled
     os: linux
-    dist: trusty
     addons:
       apt:
-        sources:
-        - sourceline: 'ppa:ubuntu-toolchain-r/test'
         packages:
-        - g++-6
+        - libsodium-dev
-        - gcc-6
   - env: BUILD_TYPE=android CURVE=tweetnacl
     os: linux
     dist: trusty
@@ -113,9 +109,6 @@ matrix:
     os: linux
     addons:
       apt:
-        sources:
-        - llvm-toolchain-trusty-8
-        - ubuntu-toolchain-r-test
         packages:
         - clang-format-8
   - env: BUILD_TYPE=default POLLER=poll

Makefile.am

@@ -1186,22 +1186,30 @@ install-data-hook:
 	$(LN_S) -r -f $(DESTDIR)/$(FUZZINGdir)/zmtp.dict $(DESTDIR)/$(FUZZINGdir)/test_bind_null_fuzzer.dict
 	$(LN_S) -r -f $(DESTDIR)/$(FUZZINGdir)/zmtp.dict $(DESTDIR)/$(FUZZINGdir)/test_connect_curve_fuzzer.dict
 	$(LN_S) -r -f $(DESTDIR)/$(FUZZINGdir)/zmtp.dict $(DESTDIR)/$(FUZZINGdir)/test_connect_null_fuzzer.dict
-	$(shell cat $(DESTDIR)/$(FUZZINGdir)/test_bind_curve_fuzzer.txt | perl -e 'print pack "H*", <STDIN>' > $(DESTDIR)/$(FUZZINGdir)/test_bind_curve_fuzzer.seed; \
+	$(shell while read -r test; do \
+		echo -n $$test | perl -e 'print pack "H*", <STDIN>' > $(DESTDIR)/$(FUZZINGdir)/test_bind_curve_fuzzer.seed; \
 		export fn=$$(cat $(DESTDIR)/$(FUZZINGdir)/test_bind_curve_fuzzer.seed | sha1sum | awk '{print $$1}'); \
 		mv $(DESTDIR)/$(FUZZINGdir)/test_bind_curve_fuzzer.seed $(DESTDIR)/$(FUZZINGdir)/$$fn; \
-		zip -j -m --quiet $(DESTDIR)/$(FUZZINGdir)/test_bind_curve_fuzzer_seed_corpus.zip $(DESTDIR)/$(FUZZINGdir)/$$fn)
+		zip -j -m -g --quiet $(DESTDIR)/$(FUZZINGdir)/test_bind_curve_fuzzer_seed_corpus.zip $(DESTDIR)/$(FUZZINGdir)/$$fn; \
-	$(shell cat $(DESTDIR)/$(FUZZINGdir)/test_bind_null_fuzzer.txt | perl -e 'print pack "H*", <STDIN>' > $(DESTDIR)/$(FUZZINGdir)/test_bind_null_fuzzer.seed; \
+		done < $(DESTDIR)/$(FUZZINGdir)/test_bind_curve_fuzzer.txt)
+	$(shell while read -r test; do \
+		echo -n $$test | perl -e 'print pack "H*", <STDIN>' > $(DESTDIR)/$(FUZZINGdir)/test_bind_null_fuzzer.seed; \
 		export fn=$$(cat $(DESTDIR)/$(FUZZINGdir)/test_bind_null_fuzzer.seed | sha1sum | awk '{print $$1}'); \
 		mv $(DESTDIR)/$(FUZZINGdir)/test_bind_null_fuzzer.seed $(DESTDIR)/$(FUZZINGdir)/$$fn; \
-		zip -j -m --quiet $(DESTDIR)/$(FUZZINGdir)/test_bind_null_fuzzer_seed_corpus.zip $(DESTDIR)/$(FUZZINGdir)/$$fn)
+		zip -j -m -g --quiet $(DESTDIR)/$(FUZZINGdir)/test_bind_null_fuzzer_seed_corpus.zip $(DESTDIR)/$(FUZZINGdir)/$$fn; \
-	$(shell cat $(DESTDIR)/$(FUZZINGdir)/test_connect_curve_fuzzer.txt | perl -e 'print pack "H*", <STDIN>' > $(DESTDIR)/$(FUZZINGdir)/test_connect_curve_fuzzer.seed; \
+		done < $(DESTDIR)/$(FUZZINGdir)/test_bind_null_fuzzer.txt)
+	$(shell while read -r test; do \
+		echo -n $$test | perl -e 'print pack "H*", <STDIN>' > $(DESTDIR)/$(FUZZINGdir)/test_connect_curve_fuzzer.seed; \
 		export fn=$$(cat $(DESTDIR)/$(FUZZINGdir)/test_connect_curve_fuzzer.seed | sha1sum | awk '{print $$1}'); \
 		mv $(DESTDIR)/$(FUZZINGdir)/test_connect_curve_fuzzer.seed $(DESTDIR)/$(FUZZINGdir)/$$fn; \
-		zip -j -m --quiet $(DESTDIR)/$(FUZZINGdir)/test_connect_curve_fuzzer_seed_corpus.zip $(DESTDIR)/$(FUZZINGdir)/$$fn)
+		zip -j -m -g --quiet $(DESTDIR)/$(FUZZINGdir)/test_connect_curve_fuzzer_seed_corpus.zip $(DESTDIR)/$(FUZZINGdir)/$$fn; \
-	$(shell cat $(DESTDIR)/$(FUZZINGdir)/test_connect_null_fuzzer.txt | perl -e 'print pack "H*", <STDIN>' > $(DESTDIR)/$(FUZZINGdir)/test_connect_null_fuzzer.seed; \
+		done < $(DESTDIR)/$(FUZZINGdir)/test_connect_curve_fuzzer.txt)
+	$(shell while read -r test; do \
+		echo -n $$test | perl -e 'print pack "H*", <STDIN>' > $(DESTDIR)/$(FUZZINGdir)/test_connect_null_fuzzer.seed; \
 		export fn=$$(cat $(DESTDIR)/$(FUZZINGdir)/test_connect_null_fuzzer.seed | sha1sum | awk '{print $$1}'); \
 		mv $(DESTDIR)/$(FUZZINGdir)/test_connect_null_fuzzer.seed $(DESTDIR)/$(FUZZINGdir)/$$fn; \
-		zip -j -m --quiet $(DESTDIR)/$(FUZZINGdir)/test_connect_null_fuzzer_seed_corpus.zip $(DESTDIR)/$(FUZZINGdir)/$$fn)
+		zip -j -m -g --quiet $(DESTDIR)/$(FUZZINGdir)/test_connect_null_fuzzer_seed_corpus.zip $(DESTDIR)/$(FUZZINGdir)/$$fn; \
+		done < $(DESTDIR)/$(FUZZINGdir)/test_connect_null_fuzzer.txt)
 	rm -f $(DESTDIR)/$(FUZZINGdir)/*.txt
 else
 test_apps += tests/test_bind_null_fuzzer \
@@ -1353,6 +1361,9 @@ EXTRA_DIST = \
 	src/version.rc.in \
 	tests/CMakeLists.txt \
         tests/test_pair_tcp_cap_net_admin.cpp \
+	tests/fuzzer_corpora/endpoint.dict tests/fuzzer_corpora/zmtp.dict \
+	tests/fuzzer_corpora/test_bind_curve_fuzzer.txt tests/fuzzer_corpora/test_bind_null_fuzzer.txt \
+	tests/fuzzer_corpora/test_connect_curve_fuzzer.txt tests/fuzzer_corpora/test_connect_null_fuzzer.txt \
 	unittests/CMakeLists.txt \
         tools/curve_keygen.cpp
 

ci_build.sh

@@ -3,7 +3,7 @@
 set -x
 set -e
 
-if [ $BUILD_TYPE == "default" ]; then
+if [ $BUILD_TYPE = "default" ]; then
     mkdir tmp
     BUILD_PREFIX=$PWD/tmp
 
@@ -14,23 +14,21 @@ if [ $BUILD_TYPE == "default" ]; then
     CONFIG_OPTS+=("LDFLAGS=-L${BUILD_PREFIX}/lib")
     CONFIG_OPTS+=("PKG_CONFIG_PATH=${BUILD_PREFIX}/lib/pkgconfig")
     CONFIG_OPTS+=("--prefix=${BUILD_PREFIX}")
+    CHECK="distcheck"
 
-    if [ -n "$ADDRESS_SANITIZER" ] && [ "$ADDRESS_SANITIZER" == "enabled" ]; then
+    if [ -n "$ADDRESS_SANITIZER" ] && [ "$ADDRESS_SANITIZER" = "enabled" ]; then
         CONFIG_OPTS+=("--enable-address-sanitizer=yes")
-        CONFIG_OPTS+=("CXX=g++-6")
+        # distcheck does an out-of-tree build, and the fuzzer tests use a hard-coded relative path for simplicity
-        CONFIG_OPTS+=("CC=gcc-6")
+        CHECK="check"
-        # workaround for linker problem with ASAN options in GCC
-        # http://stackoverflow.com/questions/37603238/fsanitize-not-using-gold-linker-in-gcc-6-1
-        CONFIG_OPTS+=("LDFLAGS=-fuse-ld=gold")
     fi
 
-    if [ $USE_NSS == "yes" ]; then
+    if [ "$USE_NSS" = "yes" ]; then
         CONFIG_OPTS+=("--with-nss")
     fi
 
-    if [ -z $CURVE ]; then
+    if [ -z "$CURVE" ]; then
         CONFIG_OPTS+=("--disable-curve")
-    elif [ $CURVE == "libsodium" ]; then
+    elif [ "$CURVE" = "libsodium" ]; then
         CONFIG_OPTS+=("--with-libsodium=yes")
 
         if ! ((command -v dpkg-query >/dev/null 2>&1 && dpkg-query --list libsodium-dev >/dev/null 2>&1) || \
@@ -40,19 +38,19 @@ if [ $BUILD_TYPE == "default" ]; then
         fi
     fi
 
-    if [ -n "$GSSAPI" ] && [ "$GSSAPI" == "enabled" ]; then
+    if [ -n "$GSSAPI" ] && [ "$GSSAPI" = "enabled" ]; then
         CONFIG_OPTS+=("--with-libgssapi_krb5=yes")
     fi
 
-    if [ -n "$PGM" ] && [ "$PGM" == "enabled" ]; then
+    if [ -n "$PGM" ] && [ "$PGM" = "enabled" ]; then
         CONFIG_OPTS+=("--with-pgm=yes")
     fi
 
-    if [ -n "$NORM" ] && [ "$NORM" == "enabled" ]; then
+    if [ -n "$NORM" ] && [ "$NORM" = "enabled" ]; then
         CONFIG_OPTS+=("--with-norm=yes")
     fi
 
-    if [ -n "$TIPC" ] && [ "$TIPC" == "enabled" ]; then
+    if [ -n "$TIPC" ] && [ "$TIPC" = "enabled" ]; then
         sudo modprobe tipc
     fi
 
@@ -60,17 +58,17 @@ if [ $BUILD_TYPE == "default" ]; then
         CONFIG_OPTS+=("--with-poller=${POLLER}")
     fi
 
-    if [ -n "$TLS" ] && [ "$TLS" == "enabled" ]; then
+    if [ -n "$TLS" ] && [ "$TLS" = "enabled" ]; then
         CONFIG_OPTS+=("--with-tls=yes")
     fi
 
-    if [ -z $DRAFT ] || [ $DRAFT == "disabled" ]; then
+    if [ -z "$DRAFT" ] || [ "$DRAFT" = "disabled" ]; then
         CONFIG_OPTS+=("--enable-drafts=no")
-    elif [ $DRAFT == "enabled" ]; then
+    elif [ "$DRAFT" = "enabled" ]; then
         CONFIG_OPTS+=("--enable-drafts=yes")
     fi
 
-    if [ -n "$FORCE_98" ] && [ "$FORCE_98" == "enabled" ]; then
+    if [ -n "$FORCE_98" ] && [ "$FORCE_98" = "enabled" ]; then
         CONFIG_OPTS+=("--enable-force-CXX98-compat=yes")
     fi
 
@@ -79,7 +77,7 @@ if [ $BUILD_TYPE == "default" ]; then
         ./autogen.sh &&
         ./configure "${CONFIG_OPTS[@]}" &&
         export DISTCHECK_CONFIGURE_FLAGS="${CONFIG_OPTS[@]}" &&
-        make VERBOSE=1 -j5 distcheck
+        make VERBOSE=1 -j5 ${CHECK}
     ) || exit 1
 else
     cd ./builds/${BUILD_TYPE} && ./ci_build.sh

configure.ac

@@ -159,8 +159,8 @@ AC_ARG_ENABLE(address-sanitizer, [AS_HELP_STRING([--enable-address-sanitizer=yes
                   [ZMQ_ASAN="$enableval"])
 
 if test "x${ZMQ_ASAN}" = "xyes"; then
-    CFLAGS="${CFLAGS} -fsanitize=address"
+    CFLAGS="${CFLAGS} -fsanitize=address -fsanitize=undefined -fno-sanitize-recover=all"
-    CXXFLAGS="${CXXFLAGS} -fsanitize=address"
+    CXXFLAGS="${CXXFLAGS} -fsanitize=address -fsanitize=undefined -fno-sanitize-recover=all"
 
     AM_CONDITIONAL(ENABLE_ASAN, true)
     AC_MSG_RESULT([yes])

src/session_base.cpp

@@ -596,61 +596,6 @@ void zmq::session_base_t::reconnect ()
         _pipe->hiccup ();
 }
 
-zmq::session_base_t::connecter_factory_entry_t
-  zmq::session_base_t::_connecter_factories[] = {
-    connecter_factory_entry_t (protocol_name::tcp,
-                               &zmq::session_base_t::create_connecter_tcp),
-#ifdef ZMQ_HAVE_WS
-    connecter_factory_entry_t (protocol_name::ws,
-                               &zmq::session_base_t::create_connecter_ws),
-#endif
-#ifdef ZMQ_HAVE_WSS
-    connecter_factory_entry_t (protocol_name::wss,
-                               &zmq::session_base_t::create_connecter_wss),
-#endif
-#if defined ZMQ_HAVE_IPC
-    connecter_factory_entry_t (protocol_name::ipc,
-                               &zmq::session_base_t::create_connecter_ipc),
-#endif
-#if defined ZMQ_HAVE_TIPC
-    connecter_factory_entry_t (protocol_name::tipc,
-                               &zmq::session_base_t::create_connecter_tipc),
-#endif
-#if defined ZMQ_HAVE_VMCI
-    connecter_factory_entry_t (protocol_name::vmci,
-                               &zmq::session_base_t::create_connecter_vmci),
-#endif
-};
-
-zmq::session_base_t::connecter_factory_map_t
-  zmq::session_base_t::_connecter_factories_map (
-    _connecter_factories,
-    _connecter_factories
-      + sizeof (_connecter_factories) / sizeof (_connecter_factories[0]));
-
-zmq::session_base_t::start_connecting_entry_t
-  zmq::session_base_t::_start_connecting_entries[] = {
-    start_connecting_entry_t (protocol_name::udp,
-                              &zmq::session_base_t::start_connecting_udp),
-#if defined ZMQ_HAVE_OPENPGM
-    start_connecting_entry_t (protocol_name::pgm,
-                              &zmq::session_base_t::start_connecting_pgm),
-    start_connecting_entry_t (protocol_name::epgm,
-                              &zmq::session_base_t::start_connecting_pgm),
-#endif
-#if defined ZMQ_HAVE_NORM
-    start_connecting_entry_t (protocol_name::norm,
-                              &zmq::session_base_t::start_connecting_norm),
-#endif
-};
-
-zmq::session_base_t::start_connecting_map_t
-  zmq::session_base_t::_start_connecting_map (
-    _start_connecting_entries,
-    _start_connecting_entries
-      + sizeof (_start_connecting_entries)
-          / sizeof (_start_connecting_entries[0]));
-
 void zmq::session_base_t::start_connecting (bool wait_)
 {
     zmq_assert (_active);
@@ -661,174 +606,164 @@ void zmq::session_base_t::start_connecting (bool wait_)
     zmq_assert (io_thread);
 
     //  Create the connecter object.
-    const connecter_factory_map_t::const_iterator connecter_factories_it =
+    own_t *connecter = NULL;
-      _connecter_factories_map.find (_addr->protocol);
+    if (_addr->protocol == protocol_name::tcp) {
-    if (connecter_factories_it != _connecter_factories_map.end ()) {
+        if (!options.socks_proxy_address.empty ()) {
-        own_t *connecter =
+            address_t *proxy_address = new (std::nothrow)
-          (this->*connecter_factories_it->second) (io_thread, wait_);
+              address_t (protocol_name::tcp, options.socks_proxy_address,
-
+                         this->get_ctx ());
+            alloc_assert (proxy_address);
+            connecter = new (std::nothrow) socks_connecter_t (
+              io_thread, this, options, _addr, proxy_address, wait_);
+            alloc_assert (connecter);
+            if (!options.socks_proxy_username.empty ()) {
+                reinterpret_cast<socks_connecter_t *> (connecter)
+                  ->set_auth_method_basic (options.socks_proxy_username,
+                                           options.socks_proxy_password);
+            }
+        } else {
+            connecter = new (std::nothrow)
+              tcp_connecter_t (io_thread, this, options, _addr, wait_);
+        }
+    }
+#if defined ZMQ_HAVE_IPC
+    else if (_addr->protocol == protocol_name::ipc) {
+        connecter = new (std::nothrow)
+          ipc_connecter_t (io_thread, this, options, _addr, wait_);
+    }
+#endif
+#if defined ZMQ_HAVE_TIPC
+    else if (_addr->protocol == protocol_name::tipc) {
+        connecter = new (std::nothrow)
+          tipc_connecter_t (io_thread, this, options, _addr, wait_);
+    }
+#endif
+#if defined ZMQ_HAVE_VMCI
+    else if (_addr->protocol == protocol_name::vmci) {
+        connecter = new (std::nothrow)
+          vmci_connecter_t (io_thread, this, options, _addr, wait_);
+    }
+#endif
+#if defined ZMQ_HAVE_WS
+    else if (_addr->protocol == protocol_name::ws) {
+        connecter = new (std::nothrow) ws_connecter_t (
+          io_thread, this, options, _addr, wait_, false, std::string ());
+    }
+#endif
+#if defined ZMQ_HAVE_WSS
+    else if (_addr->protocol == protocol_name::wss) {
+        connecter = new (std::nothrow) ws_connecter_t (
+          io_thread, this, options, _addr, wait_, true, _wss_hostname);
+    }
+#endif
+    if (connecter != NULL) {
         alloc_assert (connecter);
         launch_child (connecter);
         return;
     }
-    const start_connecting_map_t::const_iterator start_connecting_it =
+
-      _start_connecting_map.find (_addr->protocol);
+    if (_addr->protocol == protocol_name::udp) {
-    if (start_connecting_it != _start_connecting_map.end ()) {
+        zmq_assert (options.type == ZMQ_DISH || options.type == ZMQ_RADIO
-        (this->*start_connecting_it->second) (io_thread);
+                    || options.type == ZMQ_DGRAM);
+
+        udp_engine_t *engine = new (std::nothrow) udp_engine_t (options);
+        alloc_assert (engine);
+
+        bool recv = false;
+        bool send = false;
+
+        if (options.type == ZMQ_RADIO) {
+            send = true;
+            recv = false;
+        } else if (options.type == ZMQ_DISH) {
+            send = false;
+            recv = true;
+        } else if (options.type == ZMQ_DGRAM) {
+            send = true;
+            recv = true;
+        }
+
+        int rc = engine->init (_addr, send, recv);
+        errno_assert (rc == 0);
+
+        send_attach (this, engine);
+
         return;
     }
 
-    zmq_assert (false);
-}
-
-#if defined ZMQ_HAVE_VMCI
-zmq::own_t *zmq::session_base_t::create_connecter_vmci (io_thread_t *io_thread_,
-                                                        bool wait_)
-{
-    return new (std::nothrow)
-      vmci_connecter_t (io_thread_, this, options, _addr, wait_);
-}
-#endif
-
-#if defined ZMQ_HAVE_TIPC
-zmq::own_t *zmq::session_base_t::create_connecter_tipc (io_thread_t *io_thread_,
-                                                        bool wait_)
-{
-    return new (std::nothrow)
-      tipc_connecter_t (io_thread_, this, options, _addr, wait_);
-}
-#endif
-
-#if defined ZMQ_HAVE_IPC
-zmq::own_t *zmq::session_base_t::create_connecter_ipc (io_thread_t *io_thread_,
-                                                       bool wait_)
-{
-    return new (std::nothrow)
-      ipc_connecter_t (io_thread_, this, options, _addr, wait_);
-}
-#endif
-
-zmq::own_t *zmq::session_base_t::create_connecter_tcp (io_thread_t *io_thread_,
-                                                       bool wait_)
-{
-    if (!options.socks_proxy_address.empty ()) {
-        address_t *proxy_address = new (std::nothrow) address_t (
-          protocol_name::tcp, options.socks_proxy_address, this->get_ctx ());
-        alloc_assert (proxy_address);
-        socks_connecter_t *connecter = new (std::nothrow) socks_connecter_t (
-          io_thread_, this, options, _addr, proxy_address, wait_);
-        alloc_assert (connecter);
-        if (!options.socks_proxy_username.empty ()) {
-            connecter->set_auth_method_basic (options.socks_proxy_username,
-                                              options.socks_proxy_password);
-        }
-        return connecter;
-    }
-    return new (std::nothrow)
-      tcp_connecter_t (io_thread_, this, options, _addr, wait_);
-}
-
-#ifdef ZMQ_HAVE_WS
-zmq::own_t *zmq::session_base_t::create_connecter_ws (io_thread_t *io_thread_,
-                                                      bool wait_)
-{
-    return new (std::nothrow) ws_connecter_t (io_thread_, this, options, _addr,
-                                              wait_, false, std::string ());
-}
-#endif
-
-#ifdef ZMQ_HAVE_WSS
-zmq::own_t *zmq::session_base_t::create_connecter_wss (io_thread_t *io_thread_,
-                                                       bool wait_)
-{
-    return new (std::nothrow) ws_connecter_t (io_thread_, this, options, _addr,
-                                              wait_, true, _wss_hostname);
-}
-#endif
-
 #ifdef ZMQ_HAVE_OPENPGM
-void zmq::session_base_t::start_connecting_pgm (io_thread_t *io_thread_)
-{
-    zmq_assert (options.type == ZMQ_PUB || options.type == ZMQ_XPUB
-                || options.type == ZMQ_SUB || options.type == ZMQ_XSUB);
 
-    //  For EPGM transport with UDP encapsulation of PGM is used.
+    //  Both PGM and EPGM transports are using the same infrastructure.
-    bool const udp_encapsulation = _addr->protocol == protocol_name::epgm;
+    if (_addr->protocol == "pgm" || _addr->protocol == "epgm") {
+        zmq_assert (options.type == ZMQ_PUB || options.type == ZMQ_XPUB
+                    || options.type == ZMQ_SUB || options.type == ZMQ_XSUB);
 
-    //  At this point we'll create message pipes to the session straight
+        //  For EPGM transport with UDP encapsulation of PGM is used.
-    //  away. There's no point in delaying it as no concept of 'connect'
+        bool const udp_encapsulation = _addr->protocol == "epgm";
-    //  exists with PGM anyway.
-    if (options.type == ZMQ_PUB || options.type == ZMQ_XPUB) {
-        //  PGM sender.
-        pgm_sender_t *pgm_sender =
-          new (std::nothrow) pgm_sender_t (io_thread_, options);
-        alloc_assert (pgm_sender);
 
-        int rc = pgm_sender->init (udp_encapsulation, _addr->address.c_str ());
+        //  At this point we'll create message pipes to the session straight
-        errno_assert (rc == 0);
+        //  away. There's no point in delaying it as no concept of 'connect'
+        //  exists with PGM anyway.
+        if (options.type == ZMQ_PUB || options.type == ZMQ_XPUB) {
+            //  PGM sender.
+            pgm_sender_t *pgm_sender =
+              new (std::nothrow) pgm_sender_t (io_thread, options);
+            alloc_assert (pgm_sender);
 
-        send_attach (this, pgm_sender);
+            int rc =
-    } else {
+              pgm_sender->init (udp_encapsulation, _addr->address.c_str ());
-        //  PGM receiver.
+            errno_assert (rc == 0);
-        pgm_receiver_t *pgm_receiver =
-          new (std::nothrow) pgm_receiver_t (io_thread_, options);
-        alloc_assert (pgm_receiver);
 
-        int rc =
+            send_attach (this, pgm_sender);
-          pgm_receiver->init (udp_encapsulation, _addr->address.c_str ());
+        } else {
-        errno_assert (rc == 0);
+            //  PGM receiver.
+            pgm_receiver_t *pgm_receiver =
+              new (std::nothrow) pgm_receiver_t (io_thread, options);
+            alloc_assert (pgm_receiver);
 
-        send_attach (this, pgm_receiver);
+            int rc =
+              pgm_receiver->init (udp_encapsulation, _addr->address.c_str ());
+            errno_assert (rc == 0);
+
+            send_attach (this, pgm_receiver);
+        }
+
+        return;
     }
-}
 #endif
 
 #ifdef ZMQ_HAVE_NORM
-void zmq::session_base_t::start_connecting_norm (io_thread_t *io_thread_)
+    if (_addr->protocol == "norm") {
-{
+        //  At this point we'll create message pipes to the session straight
-    //  At this point we'll create message pipes to the session straight
+        //  away. There's no point in delaying it as no concept of 'connect'
-    //  away. There's no point in delaying it as no concept of 'connect'
+        //  exists with NORM anyway.
-    //  exists with NORM anyway.
+        if (options.type == ZMQ_PUB || options.type == ZMQ_XPUB) {
-    if (options.type == ZMQ_PUB || options.type == ZMQ_XPUB) {
+            //  NORM sender.
-        //  NORM sender.
+            norm_engine_t *norm_sender =
-        norm_engine_t *norm_sender =
+              new (std::nothrow) norm_engine_t (io_thread, options);
-          new (std::nothrow) norm_engine_t (io_thread_, options);
+            alloc_assert (norm_sender);
-        alloc_assert (norm_sender);
 
-        int rc = norm_sender->init (_addr->address.c_str (), true, false);
+            int rc = norm_sender->init (_addr->address.c_str (), true, false);
-        errno_assert (rc == 0);
+            errno_assert (rc == 0);
 
-        send_attach (this, norm_sender);
+            send_attach (this, norm_sender);
-    } else { // ZMQ_SUB or ZMQ_XSUB
+        } else { // ZMQ_SUB or ZMQ_XSUB
 
-        //  NORM receiver.
+            //  NORM receiver.
-        norm_engine_t *norm_receiver =
+            norm_engine_t *norm_receiver =
-          new (std::nothrow) norm_engine_t (io_thread_, options);
+              new (std::nothrow) norm_engine_t (io_thread, options);
-        alloc_assert (norm_receiver);
+            alloc_assert (norm_receiver);
 
-        int rc = norm_receiver->init (_addr->address.c_str (), false, true);
+            int rc = norm_receiver->init (_addr->address.c_str (), false, true);
-        errno_assert (rc == 0);
+            errno_assert (rc == 0);
 
-        send_attach (this, norm_receiver);
+            send_attach (this, norm_receiver);
+        }
+        return;
     }
-}
+#endif // ZMQ_HAVE_NORM
-#endif
 
-void zmq::session_base_t::start_connecting_udp (io_thread_t * /*io_thread_*/)
+    zmq_assert (false);
-{
-    zmq_assert (options.type == ZMQ_DISH || options.type == ZMQ_RADIO
-                || options.type == ZMQ_DGRAM);
-
-    udp_engine_t *engine = new (std::nothrow) udp_engine_t (options);
-    alloc_assert (engine);
-
-    const bool recv = options.type == ZMQ_DISH || options.type == ZMQ_DGRAM;
-    const bool send = options.type == ZMQ_RADIO || options.type == ZMQ_DGRAM;
-
-    const int rc = engine->init (_addr, send, recv);
-    errno_assert (rc == 0);
-
-    send_attach (this, engine);
 }
 
 zmq::hello_msg_session_t::hello_msg_session_t (io_thread_t *io_thread_,

src/session_base.hpp

@@ -106,35 +106,6 @@ class session_base_t : public own_t, public io_object_t, public i_pipe_events
   private:
     void start_connecting (bool wait_);
 
-    typedef own_t *(session_base_t::*connecter_factory_fun_t) (
-      io_thread_t *io_thread, bool wait_);
-    typedef std::pair<const std::string, connecter_factory_fun_t>
-      connecter_factory_entry_t;
-    static connecter_factory_entry_t _connecter_factories[];
-    typedef std::map<std::string, connecter_factory_fun_t>
-      connecter_factory_map_t;
-    static connecter_factory_map_t _connecter_factories_map;
-
-    own_t *create_connecter_vmci (io_thread_t *io_thread_, bool wait_);
-    own_t *create_connecter_tipc (io_thread_t *io_thread_, bool wait_);
-    own_t *create_connecter_ipc (io_thread_t *io_thread_, bool wait_);
-    own_t *create_connecter_tcp (io_thread_t *io_thread_, bool wait_);
-    own_t *create_connecter_ws (io_thread_t *io_thread_, bool wait_);
-    own_t *create_connecter_wss (io_thread_t *io_thread_, bool wait_);
-
-    typedef void (session_base_t::*start_connecting_fun_t) (
-      io_thread_t *io_thread);
-    typedef std::pair<const std::string, start_connecting_fun_t>
-      start_connecting_entry_t;
-    static start_connecting_entry_t _start_connecting_entries[];
-    typedef std::map<std::string, start_connecting_fun_t>
-      start_connecting_map_t;
-    static start_connecting_map_t _start_connecting_map;
-
-    void start_connecting_pgm (io_thread_t *io_thread_);
-    void start_connecting_norm (io_thread_t *io_thread_);
-    void start_connecting_udp (io_thread_t *io_thread_);
-
     void reconnect ();
 
     //  Handlers for incoming commands.

tests/fuzzer_corpora/test_bind_null_fuzzer.txt

@@ -1 +1,2 @@
 ff00000000000000017f03014e554c4c00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000004190552454144590b536f636b65742d5479706500000003535542040a09535542534352494245
+ff000000200000000000008585858585000000000000004700280000000000000000002100006d000028000000000000000000000000000004000019

tests/test_bind_curve_fuzzer.cpp

@@ -82,16 +82,21 @@ extern "C" int LLVMFuzzerTestOneInput (const uint8_t *data, size_t size)
 #ifndef ZMQ_USE_FUZZING_ENGINE
 void test_bind_curve_fuzzer ()
 {
-    uint8_t *data;
+    uint8_t **data;
-    size_t len;
+    size_t *len, num_cases = 0;
     if (fuzzer_corpus_encode ("tests/fuzzer_corpora/test_bind_curve_fuzzer.txt",
-                              &data, &len)
+                              &data, &len, &num_cases)
         != 0)
         exit (77);
 
-    TEST_ASSERT_SUCCESS_ERRNO (LLVMFuzzerTestOneInput (data, len));
+    while (num_cases-- > 0) {
+        TEST_ASSERT_SUCCESS_ERRNO (
+          LLVMFuzzerTestOneInput (data[num_cases], len[num_cases]));
+        free (data[num_cases]);
+    }
 
     free (data);
+    free (len);
 }
 
 int main (int argc, char **argv)

tests/test_bind_null_fuzzer.cpp

@@ -41,6 +41,11 @@ extern "C" int LLVMFuzzerTestOneInput (const uint8_t *data, size_t size)
     setup_test_context ();
     char my_endpoint[MAX_SOCKET_STRING];
     void *server = test_context_socket (ZMQ_PUB);
+    //  As per API by default there's no limit to the size of a message,
+    //  but the sanitizer allocator will barf over a gig or so
+    int64_t max_msg_size = 64 * 1024 * 1024;
+    TEST_ASSERT_SUCCESS_ERRNO (
+      zmq_setsockopt (server, ZMQ_MAXMSGSIZE, &max_msg_size, sizeof (int64_t)));
     bind_loopback_ipv4 (server, my_endpoint, sizeof (my_endpoint));
     fd_t client = connect_socket (my_endpoint);
 
@@ -70,21 +75,26 @@ extern "C" int LLVMFuzzerTestOneInput (const uint8_t *data, size_t size)
 #ifndef ZMQ_USE_FUZZING_ENGINE
 void test_bind_null_fuzzer ()
 {
-    uint8_t *data;
+    uint8_t **data;
-    size_t len;
+    size_t *len, num_cases = 0;
     if (fuzzer_corpus_encode ("tests/fuzzer_corpora/test_bind_null_fuzzer.txt",
-                              &data, &len)
+                              &data, &len, &num_cases)
         != 0)
         exit (77);
 
-    TEST_ASSERT_SUCCESS_ERRNO (LLVMFuzzerTestOneInput (data, len));
+    while (num_cases-- > 0) {
+        TEST_ASSERT_SUCCESS_ERRNO (
+          LLVMFuzzerTestOneInput (data[num_cases], len[num_cases]));
+        free (data[num_cases]);
+    }
 
     free (data);
+    free (len);
 }
 
 int main (int argc, char **argv)
 {
-    setup_test_environment ();
+    setup_test_environment (0);
 
     UNITY_BEGIN ();
     RUN_TEST (test_bind_null_fuzzer);

tests/test_connect_curve_fuzzer.cpp

@@ -86,16 +86,22 @@ extern "C" int LLVMFuzzerTestOneInput (const uint8_t *data, size_t size)
 #ifndef ZMQ_USE_FUZZING_ENGINE
 void test_connect_curve_fuzzer ()
 {
-    uint8_t *data;
+    uint8_t **data;
-    size_t len;
+    size_t *len, num_cases = 0;
     if (fuzzer_corpus_encode (
-          "tests/fuzzer_corpora/test_connect_curve_fuzzer.txt", &data, &len)
+          "tests/fuzzer_corpora/test_connect_curve_fuzzer.txt", &data, &len,
+          &num_cases)
         != 0)
         exit (77);
 
-    TEST_ASSERT_SUCCESS_ERRNO (LLVMFuzzerTestOneInput (data, len));
+    while (num_cases-- > 0) {
+        TEST_ASSERT_SUCCESS_ERRNO (
+          LLVMFuzzerTestOneInput (data[num_cases], len[num_cases]));
+        free (data[num_cases]);
+    }
 
     free (data);
+    free (len);
 }
 
 int main (int argc, char **argv)

tests/test_connect_null_fuzzer.cpp

@@ -43,6 +43,11 @@ extern "C" int LLVMFuzzerTestOneInput (const uint8_t *data, size_t size)
     fd_t server = bind_socket_resolve_port ("127.0.0.1", "0", my_endpoint);
 
     void *client = test_context_socket (ZMQ_SUB);
+    //  As per API by default there's no limit to the size of a message,
+    //  but the sanitizer allocator will barf over a gig or so
+    int64_t max_msg_size = 64 * 1024 * 1024;
+    TEST_ASSERT_SUCCESS_ERRNO (
+      zmq_setsockopt (client, ZMQ_MAXMSGSIZE, &max_msg_size, sizeof (int64_t)));
     TEST_ASSERT_SUCCESS_ERRNO (zmq_setsockopt (client, ZMQ_SUBSCRIBE, "", 0));
     TEST_ASSERT_SUCCESS_ERRNO (zmq_connect (client, my_endpoint));
 
@@ -81,16 +86,22 @@ extern "C" int LLVMFuzzerTestOneInput (const uint8_t *data, size_t size)
 #ifndef ZMQ_USE_FUZZING_ENGINE
 void test_connect_null_fuzzer ()
 {
-    uint8_t *data;
+    uint8_t **data;
-    size_t len;
+    size_t *len, num_cases = 0;
     if (fuzzer_corpus_encode (
-          "tests/fuzzer_corpora/test_connect_null_fuzzer.txt", &data, &len)
+          "tests/fuzzer_corpora/test_connect_null_fuzzer.txt", &data, &len,
+          &num_cases)
         != 0)
         exit (77);
 
-    TEST_ASSERT_SUCCESS_ERRNO (LLVMFuzzerTestOneInput (data, len));
+    while (num_cases-- > 0) {
+        TEST_ASSERT_SUCCESS_ERRNO (
+          LLVMFuzzerTestOneInput (data[num_cases], len[num_cases]));
+        free (data[num_cases]);
+    }
 
     free (data);
+    free (len);
 }
 
 int main (int argc, char **argv)

tests/testutil.cpp

@@ -523,7 +523,10 @@ bool strneq (const char *lhs_, const char *rhs_)
     return strcmp (lhs_, rhs_) != 0;
 }
 
-int fuzzer_corpus_encode (const char *filename, uint8_t **data, size_t *len)
+int fuzzer_corpus_encode (const char *filename,
+                          uint8_t ***data,
+                          size_t **len,
+                          size_t *num_cases)
 {
     TEST_ASSERT_NOT_NULL (filename);
     TEST_ASSERT_NOT_NULL (data);
@@ -532,24 +535,38 @@ int fuzzer_corpus_encode (const char *filename, uint8_t **data, size_t *len)
     if (!f)
         return -1;
     fseek (f, 0, SEEK_END);
-    size_t text_len = ftell (f);
+    size_t text_len = ftell (f) + 1;
     fseek (f, 0, SEEK_SET);
     char *buf = (char *) malloc (text_len);
     TEST_ASSERT_NOT_NULL (buf);
-    size_t read = fread (buf, 1, text_len, f);
-    fclose (f);
-    TEST_ASSERT_EQUAL_INT (read, text_len);
 
+    *len = NULL;
+    *data = NULL;
+    *num_cases = 0;
     //  Convert to binary format, corpus is stored in ascii (hex)
-    *len = text_len / 2;
+    while (fgets (buf, (int) text_len, f)) {
-    *data = (unsigned char *) malloc (*len);
+        *len = (size_t *) realloc (*len, (*num_cases + 1) * sizeof (size_t));
-    TEST_ASSERT_NOT_NULL (*data);
+        TEST_ASSERT_NOT_NULL (*len);
-    const char *pos = buf;
+        *(*len + *num_cases) = strlen (buf) / 2;
-    for (size_t count = 0; count < *len; ++count, pos += 2) {
+        *data =
-        char tmp[3] = {pos[0], pos[1], 0};
+          (uint8_t **) realloc (*data, (*num_cases + 1) * sizeof (uint8_t *));
-        (*data)[count] = (uint8_t) strtol (tmp, NULL, 16);
+        TEST_ASSERT_NOT_NULL (*data);
+        *(*data + *num_cases) =
+          (uint8_t *) malloc (*(*len + *num_cases) * sizeof (uint8_t));
+        TEST_ASSERT_NOT_NULL (*(*data + *num_cases));
+
+        const char *pos = buf;
+        for (size_t count = 0; count < *(*len + *num_cases);
+             ++count, pos += 2) {
+            char tmp[3] = {pos[0], pos[1], 0};
+            *(*(*data + *num_cases) + count) = (uint8_t) strtol (tmp, NULL, 16);
+        }
+        (*num_cases)++;
     }
+
+
     free (buf);
+    fclose (f);
 
     return 0;
 }

tests/testutil.hpp

@@ -228,6 +228,9 @@ fd_t bind_socket_resolve_port (const char *address_,
                                const int af_ = AF_INET,
                                const int protocol_ = IPPROTO_TCP);
 
-int fuzzer_corpus_encode (const char *filename, uint8_t **data, size_t *len);
+int fuzzer_corpus_encode (const char *filename,
+                          uint8_t ***data,
+                          size_t **len,
+                          size_t *num_cases);
 
 #endif

tests/testutil_security.cpp

@@ -320,6 +320,11 @@ void setup_context_and_server_side (void **zap_control_,
     *server_ = test_context_socket (ZMQ_DEALER);
     TEST_ASSERT_SUCCESS_ERRNO (
       zmq_setsockopt (*server_, ZMQ_LINGER, &linger, sizeof (linger)));
+    //  As per API by default there's no limit to the size of a message,
+    //  but the sanitizer allocator will barf over a gig or so
+    int64_t max_msg_size = 64 * 1024 * 1024;
+    TEST_ASSERT_SUCCESS_ERRNO (zmq_setsockopt (
+      *server_, ZMQ_MAXMSGSIZE, &max_msg_size, sizeof (int64_t)));
 
     socket_config_ (*server_, socket_config_data_);
 
@@ -363,6 +368,11 @@ void *create_and_connect_client (char *my_endpoint_,
                                  void **client_mon_)
 {
     void *client = test_context_socket (ZMQ_DEALER);
+    //  As per API by default there's no limit to the size of a message,
+    //  but the sanitizer allocator will barf over a gig or so
+    int64_t max_msg_size = 64 * 1024 * 1024;
+    TEST_ASSERT_SUCCESS_ERRNO (
+      zmq_setsockopt (client, ZMQ_MAXMSGSIZE, &max_msg_size, sizeof (int64_t)));
 
     socket_config_ (client, socket_config_data_);