Fixes #1753: Expire Listener-side connection on certificate rotation (#1763)

* Fixes #1753: Expire Listener-side connection on certificate rotation

When an AMQP listener's sslProfile oldestValidOrdinal attribute is
updated scan all connections that are children of that listener and
check if the connection's tlsOrdinal value is less than the new
oldestValidOrdinal value. If it is then close the connection.

Patch includes additional efforts to make the code fluffy:

1) Move qd_listener_t-specific code out of connection_manager.c and
move it to the qd_listener.c file.
2) Modify the qd_connection_t deferred callback API to allow an extra
context parameter to be passed to the callback. Hide the
implementation details by moving the data structure from a header file
into qd_connection.c.
3) Remove dead code.

* fixup: verify tlsOrdinal of new inter-router connections

Author: kgiusti

include/qpid/dispatch/server.h

@@ -43,19 +43,6 @@ typedef struct sys_mutex_t      sys_mutex_t;
  * @{
  */
 
-/**
- * Deferred callback
- *
- * This type is for calls that are deferred until they can be invoked on
- * a specific connection's thread.
- *
- * @param context An opaque context to be passed back with the call.
- * @param discard If true, the call should be discarded because the connection it
- *        was pending on was deleted.
- */
-typedef void (*qd_deferred_t)(void *context, bool discard);
-
-
 /**
  * Run the server threads until completion - The blocking version.
  *

src/adaptors/amqp/amqp_adaptor.c

@@ -57,7 +57,7 @@ static char *edge_role        = "edge";
 static char *inter_edge_role  = "inter-edge";
 
 
-static void deferred_AMQP_rx_handler(void *context, bool discard);
+static void deferred_AMQP_rx_handler(qd_connection_t *qd_conn, void *context, bool discard);
 static bool parse_failover_property_list(qd_router_t *router, qd_connection_t *conn, pn_data_t *props);
 static void clear_producer_activation(qdr_core_t *core, qdr_delivery_t *delivery);
 static void clear_consumer_activation(qdr_core_t *core, qdr_delivery_t *delivery);
@@ -980,7 +980,7 @@ bool AMQP_rx_handler(qd_router_t *router, qd_link_t *link)
 /**
  * Deferred callback for inbound delivery handler
  */
-static void deferred_AMQP_rx_handler(void *context, bool discard)
+static void deferred_AMQP_rx_handler(qd_connection_t *qd_conn, void *context, bool discard)
 {
     qd_link_t_sp *safe_qdl = (qd_link_t_sp*) context;
 

src/adaptors/amqp/connection_manager.c

@@ -25,7 +25,6 @@
 #include "server_config.h"
 #include "dispatch_private.h"
 #include "entity.h"
-#include "server_private.h"
 
 #include "qpid/dispatch/ctools.h"
 #include "qpid/dispatch/failoverlist.h"
@@ -42,7 +41,6 @@
 
 
 struct qd_connection_manager_t {
-    qd_server_t                  *server;
     qd_listener_list_t            listeners;
     qd_connector_config_list_t    connector_configs;
 };
@@ -60,58 +58,11 @@ static void log_config(qd_server_config_t *c, const char *what, bool create)
 QD_EXPORT qd_listener_t *qd_dispatch_configure_listener(qd_dispatch_t *qd, qd_entity_t *entity)
 {
     qd_connection_manager_t *cm = qd->connection_manager;
-    qd_listener_t *li = qd_listener(qd->server);
-    if (!li || qd_server_config_load(&li->config, entity, true) != QD_ERROR_NONE) {
-        qd_log(LOG_CONN_MGR, QD_LOG_ERROR, "Unable to create listener: %s", qd_error_message());
-        qd_listener_decref(li);
+    qd_listener_t *li = qd_listener_create(qd, entity);
+    if (!li) {
         return 0;
     }
 
-    if (li->config.ssl_profile_name) {
-        li->tls_config = qd_tls_config(li->config.ssl_profile_name,
-                                       QD_TLS_TYPE_PROTON_AMQP,
-                                       QD_TLS_CONFIG_SERVER_MODE,
-                                       li->config.verify_host_name,
-                                       li->config.ssl_require_peer_authentication);
-        if (!li->tls_config) {
-            // qd_tls_config() sets qd_error_message():
-            qd_log(LOG_CONN_MGR, QD_LOG_ERROR, "Failed to configure TLS for Listener %s: %s",
-                   li->config.name, qd_error_message());
-            qd_listener_decref(li);
-            return 0;
-        }
-        li->tls_ordinal              = qd_tls_config_get_ordinal(li->tls_config);
-        li->tls_oldest_valid_ordinal = qd_tls_config_get_oldest_valid_ordinal(li->tls_config);
-    }
-
-    char *fol = qd_entity_opt_string(entity, "failoverUrls", 0);
-    if (fol) {
-        li->config.failover_list = qd_failover_list(fol);
-        free(fol);
-        if (li->config.failover_list == 0) {
-            qd_log(LOG_CONN_MGR, QD_LOG_ERROR, "Unable to create listener, bad failover list: %s",
-                   qd_error_message());
-            qd_listener_decref(li);
-            return 0;
-        }
-    } else {
-        li->config.failover_list = 0;
-    }
-
-    //
-    // Set up the vanflow record for this listener (ROUTER_ACCESS).
-    // Do this only for router-to-router links: not mgmt/metrics/healthz/websockets listeners
-    //
-    if (strcmp(li->config.role, "inter-router") == 0 ||
-        strcmp(li->config.role, "edge") == 0 ||
-        strcmp(li->config.role, "inter-edge") == 0) {
-        li->vflow_record = vflow_start_record(VFLOW_RECORD_ROUTER_ACCESS, 0);
-        vflow_set_string(li->vflow_record, VFLOW_ATTRIBUTE_NAME, li->config.name);
-        vflow_set_string(li->vflow_record, VFLOW_ATTRIBUTE_ROLE, li->config.role);
-        vflow_set_uint64(li->vflow_record, VFLOW_ATTRIBUTE_LINK_COUNT, 0);
-    }
-
-    DEQ_ITEM_INIT(li);
     DEQ_INSERT_TAIL(cm->listeners, li);
     log_config(&li->config, "Listener", true);
     return li;
@@ -299,7 +250,6 @@ qd_connection_manager_t *qd_connection_manager(qd_dispatch_t *qd)
     if (!cm)
         return 0;
 
-    cm->server     = qd->server;
     DEQ_INIT(cm->listeners);
     DEQ_INIT(cm->connector_configs);
 
@@ -315,17 +265,7 @@ void qd_connection_manager_free(qd_connection_manager_t *cm)
     qd_listener_t *li = DEQ_HEAD(cm->listeners);
     while (li) {
         DEQ_REMOVE_HEAD(cm->listeners);
-        if (li->pn_listener) {
-            // DISPATCH-1508: force cleanup of pn_listener context.  This is
-            // usually done in the PN_LISTENER_CLOSE event handler in server.c,
-            // but since the router is going down those events will no longer
-            // be generated.
-            pn_listener_set_context(li->pn_listener, 0);
-            pn_listener_close(li->pn_listener);
-            li->pn_listener = 0;
-            qd_listener_decref(li);  // for the pn_listener's context
-        }
-        qd_listener_decref(li);
+        qd_listener_delete(li, true);  // true == router is shutting down
         li = DEQ_HEAD(cm->listeners);
     }
 
@@ -351,6 +291,10 @@ QD_EXPORT void qd_connection_manager_start(qd_dispatch_t *qd)
 
     while (li) {
         if (!li->pn_listener) {
+            // DISPATCH-55: failure to bind on router initialization can result in a router that cannot be accessed by
+            // management, preventing diagnosing/fixing the issue.  Treat listener failure on initial bring up as a
+            // critical issue. Failure of listeners added after the router has been successfully started will simply
+            // result in a logged error.
             if (!qd_listener_listen(li) && first_start) {
                 qd_log(LOG_CONN_MGR, QD_LOG_CRITICAL, "Listen on %s failed during initial config",
                        li->config.host_port);
@@ -375,17 +319,9 @@ QD_EXPORT void qd_connection_manager_delete_listener(qd_dispatch_t *qd, void *im
 {
     qd_listener_t *li = (qd_listener_t*) impl;
     if (li) {
-        if (li->pn_listener) {
-            pn_listener_close(li->pn_listener);
-        }
-        else if (li->http) {
-            qd_lws_listener_close(li->http);
-        }
-
-        log_config(&li->config, "Listener", false);
-
         DEQ_REMOVE(qd->connection_manager->listeners, li);
-        qd_listener_decref(li);
+        log_config(&li->config, "Listener", false);
+        qd_listener_delete(li, false);  // false == do a clean listener close
     }
 }
 

src/adaptors/amqp/qd_connection.c

@@ -38,9 +38,18 @@
 #include <inttypes.h>
 
 
-ALLOC_DEFINE(qd_deferred_call_t);
 ALLOC_DEFINE_SAFE(qd_connection_t);
 
+/**
+ * Context for the deferred callback
+ */
+struct qd_deferred_call_t {
+    DEQ_LINKS(struct qd_deferred_call_t);
+    qd_deferred_cb_t  call;
+    void             *context;
+};
+ALLOC_DEFINE(qd_deferred_call_t);
+
 const char *MECH_EXTERNAL = "EXTERNAL";
 
 
@@ -474,24 +483,16 @@ const qd_server_config_t *qd_connection_config(const qd_connection_t *conn)
 }
 
 
-void qd_connection_invoke_deferred(qd_connection_t *conn, qd_deferred_t call, void *context)
+void qd_connection_invoke_deferred(qd_connection_t *conn, qd_deferred_cb_t call, void *context)
 {
-    if (!conn)
-        return;
-
-    qd_connection_invoke_deferred_impl(conn, call, context, new_qd_deferred_call_t());
+    assert(conn);
+    qd_connection_invoke_deferred_impl(conn, qd_connection_new_qd_deferred_call_t(call, context));
 }
 
 
-void qd_connection_invoke_deferred_impl(qd_connection_t *conn, qd_deferred_t call, void *context, void *dct)
+void qd_connection_invoke_deferred_impl(qd_connection_t *conn, qd_deferred_call_t *dc)
 {
-    if (!conn)
-        return;
-
-    qd_deferred_call_t *dc = (qd_deferred_call_t*)dct;
-    DEQ_ITEM_INIT(dc);
-    dc->call    = call;
-    dc->context = context;
+    assert(!!conn && !!dc);
 
     sys_mutex_lock(&conn->deferred_call_lock);
     DEQ_INSERT_TAIL(conn->deferred_calls, dc);
@@ -502,37 +503,46 @@ void qd_connection_invoke_deferred_impl(qd_connection_t *conn, qd_deferred_t cal
     sys_mutex_unlock(qd_server_get_activation_lock(conn->server));
 }
 
-void *qd_connection_new_qd_deferred_call_t(void)
+
+qd_deferred_call_t *qd_connection_new_qd_deferred_call_t(qd_deferred_cb_t callback, void *context)
 {
-    return new_qd_deferred_call_t();
+    qd_deferred_call_t *dc = new_qd_deferred_call_t();
+    DEQ_ITEM_INIT(dc);
+    dc->call    = callback;
+    dc->context = context;
+    return dc;
 }
 
 
-void qd_connection_free_qd_deferred_call_t(void *dct)
+void qd_connection_free_qd_deferred_call_t(qd_deferred_call_t *dc)
 {
-    free_qd_deferred_call_t((qd_deferred_call_t *)dct);
+    free_qd_deferred_call_t(dc);
 }
 
+
 void qd_connection_invoke_deferred_calls(qd_connection_t *conn, bool discard)
 {
-    if (!conn)
-        return;
+    assert(conn);
 
-    // Lock access to deferred_calls, other threads may concurrently add to it.  Invoke
+    // Snapshot the deferred_calls list under lock since other threads may concurrently add to it.  Invoke
     // the calls outside of the critical section.
     //
+    qd_deferred_call_list_t deferred_calls = DEQ_EMPTY;
     sys_mutex_lock(&conn->deferred_call_lock);
-    qd_deferred_call_t *dc;
-    while ((dc = DEQ_HEAD(conn->deferred_calls))) {
-        DEQ_REMOVE_HEAD(conn->deferred_calls);
-        sys_mutex_unlock(&conn->deferred_call_lock);
-        dc->call(dc->context, discard);
-        free_qd_deferred_call_t(dc);
-        sys_mutex_lock(&conn->deferred_call_lock);
-    }
+    DEQ_MOVE(conn->deferred_calls, deferred_calls);
     sys_mutex_unlock(&conn->deferred_call_lock);
-}
 
+    qd_deferred_call_t *dc = DEQ_HEAD(deferred_calls);
+    while (dc) {
+        // Note: this destroys the list as it is traversed. This is an optimization: since everything is freed do not
+        // bother with the overhead of dequeing
+        qd_deferred_call_t *tmp = dc;
+        dc = DEQ_NEXT(dc);
+
+        tmp->call(conn, tmp->context, discard);
+        free_qd_deferred_call_t(tmp);
+    }
+}
 
 
 const char* qd_connection_name(const qd_connection_t *c)
@@ -721,15 +731,14 @@ void qd_container_handle_event(qd_container_t *container, pn_event_t *event, pn_
 void qd_conn_event_batch_complete(qd_container_t *container, qd_connection_t *qd_conn, bool conn_closed);
 
 
-static void timeout_on_handshake(void *context, bool discard)
+static void timeout_on_handshake(qd_connection_t *qd_conn, void *context, bool discard)
 {
     if (discard)
         return;
 
-    qd_connection_t *ctx   = (qd_connection_t*) context;
-    pn_transport_t  *tport = pn_connection_transport(ctx->pn_conn);
-    pn_transport_close_head(tport);
-    connect_fail(ctx, QD_AMQP_COND_NOT_ALLOWED, "Timeout waiting for initial handshake");
+    pn_transport_t  *tport = pn_connection_transport(qd_conn->pn_conn);
+    pn_transport_close_head(tport);  // force close to avoid doing handshake (may hang otherwise)
+    connect_fail(qd_conn, QD_AMQP_COND_NOT_ALLOWED, "Timeout waiting for initial handshake");
 }
 
 

src/adaptors/amqp/qd_connection.h

@@ -47,19 +47,9 @@ typedef struct qd_session_t         qd_session_t;
 typedef struct qd_timer_t           qd_timer_t;
 typedef struct qd_tls_session_t     qd_tls_session_t;
 
-
-// defer a function call to be invoked on the connections proactor thread at
-// a later time
-//
-typedef struct qd_deferred_call_t {
-    DEQ_LINKS(struct qd_deferred_call_t);
-    qd_deferred_t  call;
-    void          *context;
-} qd_deferred_call_t;
-
+typedef struct qd_deferred_call_t   qd_deferred_call_t;
 DEQ_DECLARE(qd_deferred_call_t, qd_deferred_call_list_t);
 
-
 typedef struct qd_pn_free_link_t qd_pn_free_link_t;
 DEQ_DECLARE(qd_pn_free_link_t, qd_pn_free_link_list_t);
 
@@ -146,39 +136,47 @@ const char* qd_connection_name(const qd_connection_t *c);
  * Schedule a call to be invoked on a thread that has ownership of this connection.
  * It will be safe for the callback to perform operations related to this connection.
  *
- * @param conn Connection object
- * @param call The function to be invoked on the connection's thread
+ * @param qd_conn Connection object. Note well: the caller must ensure qd_conn remains valid until after
+ *        this call returns.
+ * @param call The function to be invoked asynchronously on the connection's thread.
  * @param context The context to be passed back in the callback
  */
-void qd_connection_invoke_deferred(qd_connection_t *conn, qd_deferred_t call, void *context);
-
-void qd_connection_invoke_deferred_calls(qd_connection_t *conn, bool discard);
+typedef void (*qd_deferred_cb_t)(qd_connection_t *qd_conn, void *context, bool discard);
+void qd_connection_invoke_deferred(qd_connection_t *conn, qd_deferred_cb_t call, void *context);
 
 /**
- * Schedule a call to be invoked on a thread that has ownership of this connection
- * when it will be safe for the callback to perform operations related to this connection.
- * A qd_deferred_call_t object has been allocated before hand to avoid taking
- * the ENTITY_CACHE lock.
+ * Alternative qd_connection_invoke_deferred() API
  *
- * @param conn Connection object
- * @param call The function to be invoked on the connection's thread
- * @param context The context to be passed back in the callback
- * @param dct Pointer to preallocated qd_deferred_call_t object
+ * Allows the caller to pre-allocate the deferred callback context. This allows the context to be allocated outside of a
+ * critical section where the qd_connection_t has been locked.
+ *
+ * @param conn Connection object. Note well: the caller must ensure qd_conn remains valid until after
+ *        this call returns.
+ * @param call_context The qd_deferred_call_t allocated via qd_connection_new_qd_deferred_call_t(). Ownership is passed
+ *        to the call - do not call qd_connection_free_qd_deferred_call_t()
  */
-void qd_connection_invoke_deferred_impl(qd_connection_t *conn, qd_deferred_t call, void *context, void *dct);
-
+void qd_connection_invoke_deferred_impl(qd_connection_t *conn, qd_deferred_call_t *call_context);
 
 /**
  * Allocate a qd_deferred_call_t object
  */
-void *qd_connection_new_qd_deferred_call_t(void);
+qd_deferred_call_t *qd_connection_new_qd_deferred_call_t(qd_deferred_cb_t callback, void *context);
 
 /**
- * Deallocate a qd_deferred_call_t object
+ * Deallocate a qd_deferred_call_t object. Call this only in the case where the call context was not passed to
+ * qd_connection_invoke_deferred_impl()
  *
  * @param dct Pointer to preallocated qd_deferred_call_t object
  */
-void qd_connection_free_qd_deferred_call_t(void *dct);
+void qd_connection_free_qd_deferred_call_t(qd_deferred_call_t *dct);
+
+
+/**
+ * Run all deferred callbacks.
+ *
+ * Invoked by the PN_CONNECTION_WAKE event handler
+ */
+void qd_connection_invoke_deferred_calls(qd_connection_t *conn, bool discard);
 
 
 /**
@@ -304,5 +302,4 @@ bool qd_connection_strip_annotations_in(const qd_connection_t *c);
  * ordinal.
  */
 bool qd_connection_get_tls_ordinal(const qd_connection_t *qd_conn, uint64_t *tls_ordinal);
-
 #endif