DISPATCH-2307: prevent router failure if id is greater than 64 bytes

fixes:
  - buffer overflow in iterator library
  - incorrect fanout value for locally destined messages

This closes #1478

Author: kgiusti

include/qpid/dispatch/iterator.h

@@ -139,6 +139,12 @@ typedef struct {
  * @{
  */
 
+/**
+ * Clean up any internal state in the iterator library.  This must be called at
+ * shutdown after all iterators have been released.
+ */
+void qd_iterator_finalize(void);
+
 /**
  * Set the area and router names for the local router.  These are used to match
  * my-area and my-router in address fields.  These settings are global and used

include/qpid/dispatch/message.h

@@ -547,12 +547,10 @@ void qd_message_set_tag_sent(qd_message_t *msg, bool tag_sent);
 /**
  * Increase the fanout of the message by 1.
  *
- * @param in_msg A pointer to the inbound message.
- * @param out_msg A pointer to the outbound message or 0 if forwarding to a
- * local subscriber.
+ * @param out_msg A pointer to the message to be sent outbound or to a local
+ * subscriber.
  */
-void qd_message_add_fanout(qd_message_t *in_msg,
-                           qd_message_t *out_msg);
+void qd_message_add_fanout(qd_message_t *out_msg);
 
 /**
  * Disable the Q2-holdoff for this message.

src/dispatch.c

@@ -379,6 +379,7 @@ void qd_dispatch_free(qd_dispatch_t *qd)
     qd_python_finalize();
     qd_dispatch_set_router_id(qd, NULL);
     qd_dispatch_set_router_area(qd, NULL);
+    qd_iterator_finalize();
     free(qd->timestamp_format);
     free(qd->metadata);
 

src/iterator.c

@@ -84,8 +84,8 @@ typedef enum {
 
 
 static bool  edge_mode = false;
-static char *my_area   = "";
-static char *my_router = "";
+static char *my_area   = 0;
+static char *my_router = 0;
 
 static const char    *SEPARATORS = "./";
 
@@ -156,6 +156,7 @@ static void parse_address_view(qd_iterator_t *iter)
         }
 
         if (qd_iterator_prefix(iter, "topo/")) {
+            assert(my_area && my_router);  // ensure qd_iterator_set_address called!
             if (qd_iterator_prefix(iter, "all/") || qd_iterator_prefix(iter, my_area)) {
                 if (qd_iterator_prefix(iter, "all/")) {
                     iter->prefix = QD_ITER_HASH_PREFIX_TOPOLOGICAL;
@@ -567,19 +568,18 @@ static void qd_iterator_free_hash_segments(qd_iterator_t *iter)
 
 void qd_iterator_set_address(bool _edge_mode, const char *area, const char *router)
 {
-    static char  buf[64];
-    static char *ptr   = buf;
-    size_t area_size   = strlen(area);
-    size_t router_size = strlen(router);
+    const size_t area_size   = strlen(area);
+    const size_t router_size = strlen(router);
 
-    if (area_size + router_size + 1 >= sizeof(buf))
-        ptr = (char*) malloc(area_size + router_size + 2);
+    edge_mode = _edge_mode;
 
-    sprintf(ptr, "%s/%c%s/", area, '\0', router);
+    free(my_area);
+    my_area = qd_malloc(area_size + 2);  // include trailing '\'
+    sprintf(my_area, "%s/", area);
 
-    edge_mode = _edge_mode;
-    my_area   = ptr;
-    my_router = ptr + area_size + 2;
+    free(my_router);
+    my_router = qd_malloc(router_size + 2);
+    sprintf(my_router, "%s/", router);
 }
 
 
@@ -1143,3 +1143,14 @@ void qd_iterator_get_view_cursor(
     ptr->cursor    = iter->view_pointer.cursor;
     ptr->remaining = iter->view_pointer.remaining;
 }
+
+
+void qd_iterator_finalize(void)
+{
+    free(my_area);
+    free(my_router);
+
+    // unit tests need these zeroed
+    my_area = 0;
+    my_router = 0;
+}

src/message.c

@@ -1274,12 +1274,9 @@ void qd_message_set_discard(qd_message_t *msg, bool discard)
 
 // update the buffer reference counts for a new outgoing message
 //
-void qd_message_add_fanout(qd_message_t *in_msg,
-                           qd_message_t *out_msg)
+void qd_message_add_fanout(qd_message_t *out_msg)
 {
-    if (!out_msg)
-        return;
-
+    assert(out_msg);
     qd_message_pvt_t *msg = (qd_message_pvt_t *)out_msg;
     msg->is_fanout = true;
 

src/router_core/forwarder.c

@@ -180,7 +180,7 @@ qdr_delivery_t *qdr_forward_new_delivery_CT(qdr_core_t *core, qdr_delivery_t *in
     //
     // Add one to the message fanout. This will later be used in the qd_message_send function that sends out messages.
     //
-    qd_message_add_fanout(msg, out_dlv->msg);
+    qd_message_add_fanout(out_dlv->msg);
 
     //
     // Create peer linkage if the outgoing delivery is unsettled. This peer linkage is necessary to deal with dispositions that show up in the future.
@@ -439,7 +439,7 @@ static inline bool qdr_forward_edge_echo_CT(qdr_delivery_t *in_dlv, qdr_link_t *
  */
 static void qdr_forward_to_subscriber_CT(qdr_core_t *core, qdr_subscription_t *sub, qdr_delivery_t *in_dlv, qd_message_t *in_msg, bool receive_complete)
 {
-    qd_message_add_fanout(in_msg, 0);
+    qd_message_add_fanout(in_msg);
 
     //
     // Only if the message has been completely received, forward it to the subscription.

tests/c_unittests/helpers.hpp

@@ -305,6 +305,7 @@ class QDRMinimalEnv
         qd_log_finalize();
         qd_alloc_finalize();
         qd_entity_cache_free_entries();
+        qd_iterator_finalize();
     }
 };
 

tests/message_test.c

@@ -1167,9 +1167,9 @@ static char *test_check_stream_data_fanout(void *context)
 
     // "fan out" the message
     out_msg1 = qd_message_copy(in_msg);
-    qd_message_add_fanout(in_msg, out_msg1);
+    qd_message_add_fanout(out_msg1);
     out_msg2 = qd_message_copy(in_msg);
-    qd_message_add_fanout(in_msg, out_msg2);
+    qd_message_add_fanout(out_msg2);
 
     // walk the data streams for both messages:
     qd_message_stream_data_t *out_sd1[sd_count] = {0};
@@ -1291,9 +1291,9 @@ static char *test_check_stream_data_footer(void *context)
 
     // "fan out" the message
     out_msg1 = qd_message_copy(in_msg);
-    qd_message_add_fanout(in_msg, out_msg1);
+    qd_message_add_fanout(out_msg1);
     out_msg2 = qd_message_copy(in_msg);
-    qd_message_add_fanout(in_msg, out_msg2);
+    qd_message_add_fanout(out_msg2);
 
     qd_message_stream_data_t *stream_data = 0;
     bool done = false;

tests/run_unit_tests_size.c

@@ -19,6 +19,7 @@
 
 #include "qpid/dispatch/alloc.h"
 #include "qpid/dispatch/buffer.h"
+#include "qpid/dispatch/iterator.h"
 
 void qd_log_initialize(void);
 void qd_log_finalize(void);
@@ -52,6 +53,7 @@ int main(int argc, char** argv)
 
     qd_log_finalize();
     qd_alloc_finalize();
+    qd_iterator_finalize();
     return result;
 }
 

tests/system_tests_routing_protocol.py

@@ -204,5 +204,85 @@ def run(self):
         Container(self).run()
 
 
+class HugeRouterIdTest(TestCase):
+    """
+    Ensure long router identifiers are advertized properly.
+
+    Deploy a mesh of four routers with ids > 64 octets in length.
+    """
+    @classmethod
+    def setUpClass(cls):
+        super(HugeRouterIdTest, cls).setUpClass()
+
+        def router(name, extra_config):
+
+            config = [
+                ('router', {'mode': 'interior', 'id': name}),
+                ('listener', {'port': cls.tester.get_port()})
+            ] + extra_config
+
+            config = Qdrouterd.Config(config)
+
+            r = cls.tester.qdrouterd(name[:32], config, wait=False)
+            cls.routers.append(r)
+            return r
+
+        cls.routers = []
+
+        ir_port_A = cls.tester.get_port()
+        ir_port_B = cls.tester.get_port()
+        ir_port_C = cls.tester.get_port()
+        ir_port_D = cls.tester.get_port()
+
+        name_suffix = "." + "X" * 128
+
+        cls.RA_name = 'A' + name_suffix
+        cls.RA = router(cls.RA_name,
+                        [('listener', {'role': 'inter-router', 'port': ir_port_A}),
+                         ('connector', {'role': 'inter-router', 'port': ir_port_B}),
+                         ('connector', {'role': 'inter-router', 'port': ir_port_C}),
+                         ('connector', {'role': 'inter-router', 'port': ir_port_D})])
+
+        cls.RB_name = 'B' + name_suffix
+        cls.RB = router(cls.RB_name,
+                        [('listener', {'role': 'inter-router', 'port': ir_port_B}),
+                         ('connector', {'role': 'inter-router', 'port': ir_port_C}),
+                         ('connector', {'role': 'inter-router', 'port': ir_port_D})])
+
+        cls.RC_name = 'C' + name_suffix
+        cls.RC = router(cls.RC_name,
+                        [('listener', {'role': 'inter-router', 'port': ir_port_C}),
+                         ('connector', {'role': 'inter-router', 'port': ir_port_D})])
+
+        cls.RD_name = 'D' + name_suffix
+        cls.RD = router(cls.RD_name,
+                        [('listener', {'role': 'inter-router', 'port': ir_port_D})])
+
+        cls.RA.wait_ports()
+        cls.RB.wait_ports()
+        cls.RC.wait_ports()
+        cls.RD.wait_ports()
+
+    def test_01_wait_for_routers(self):
+        """
+        Wait until all the router in the mesh can see each other
+        """
+        self.RA.wait_router_connected(self.RB_name)
+        self.RA.wait_router_connected(self.RC_name)
+        self.RA.wait_router_connected(self.RD_name)
+
+        self.RB.wait_router_connected(self.RA_name)
+        self.RB.wait_router_connected(self.RC_name)
+        self.RB.wait_router_connected(self.RD_name)
+
+        self.RC.wait_router_connected(self.RA_name)
+        self.RC.wait_router_connected(self.RB_name)
+        self.RC.wait_router_connected(self.RD_name)
+
+        self.RD.wait_router_connected(self.RA_name)
+        self.RD.wait_router_connected(self.RB_name)
+        self.RD.wait_router_connected(self.RC_name)
+
+
 if __name__ == '__main__':
     unittest.main(main_module())