Fixes #1741: add ordinal attributes to sslProfile (#1744)

Include mechanism for propagating updates to affected connectors.
Advertise the connection's group ordinal to the peer router via the 
properties map in the Open performative.

Author: kgiusti

include/qpid/dispatch/amqp.h

@@ -155,6 +155,7 @@ extern const char * const QD_CONNECTION_PROPERTY_VERSION_KEY;
 extern const char * const QD_CONNECTION_PROPERTY_COST_KEY;
 extern const char * const QD_CONNECTION_PROPERTY_ROLE_KEY;
 extern const char * const QD_CONNECTION_PROPERTY_GROUP_CORRELATOR_KEY;
+extern const char * const QD_CONNECTION_PROPERTY_GROUP_ORDINAL_KEY;
 extern const char * const QD_CONNECTION_PROPERTY_CONN_ID;
 extern const char * const QD_CONNECTION_PROPERTY_FAILOVER_LIST_KEY;
 extern const char * const QD_CONNECTION_PROPERTY_FAILOVER_NETHOST_KEY;

include/qpid/dispatch/protocol_adaptor.h

@@ -928,13 +928,14 @@ qdr_connection_info_t *qdr_connection_info(bool             is_encrypted,
                                            const char      *user,
                                            const char      *container,
                                            pn_data_t       *connection_properties,
+                                           uint64_t         tls_ordinal,
                                            int              ssl_ssf,
                                            bool             ssl,
                                            const char      *version,
                                            bool             streaming_links,
                                            bool             connection_trunking);
 
-void qdr_connection_info_set_group_correlator(qdr_connection_info_t *info, const char *correlator);
+void qdr_connection_info_set_group(qdr_connection_info_t *info, const char *correlator, uint64_t ordinal);
 void qdr_connection_info_set_tls(qdr_connection_info_t *info, bool enabled, char *version, char *ciphers, int ssf);
 
 void qd_adaptor_listener_init(void);

include/qpid/dispatch/tls_common.h

@@ -84,15 +84,17 @@ struct qd_ssl2_profile_t {
     char *uid_name_mapping_file;
 
     /**
-     * version: Version assigned to the current configuration
-     * oldest_valid_version: Previous sslProfile updates with versions values < oldest_valid_version have expired.
+     * Certificate Rotation
+     * ordinal: Identifies the current configuration's revision
+     * oldest_valid_ordinal: Previous configurations with ordinal values < oldest_valid_ordinal have expired.
      */
-    long version;
-    long oldest_valid_version;
+    uint64_t ordinal;
+    uint64_t oldest_valid_ordinal;
 };
 
 /**
- * Create a new TLS qd_tls_config_t instance with the given configuration
+ * Create a new TLS qd_tls_config_t instance with the given configuration. This is called when a listener/connector
+ * record is created.
  *
  * @param ssl_profile_name the name of the sslProfile configuration to use
  * @param p_type protocol type for the child connections (TCP or AMQP)
@@ -117,6 +119,45 @@ qd_tls_config_t *qd_tls_config(const char *ssl_profile_name,
 void qd_tls_config_decref(qd_tls_config_t *config);
 
 
+/**
+ * Get the values of the ordinal/oldestValidOrdinal assocated with the TLS configuration.
+ *
+ * Note: To avoid races this function can only be called from the context of the management thread.
+ *
+ * @param config The qd_tls_config_t to query.
+ * @return the value for the ordinal/lastValidOrdinal sslProfile attributes used by this config
+ */
+uint64_t qd_tls_config_get_ordinal(const qd_tls_config_t *config);
+uint64_t qd_tls_config_get_oldest_valid_ordinal(const qd_tls_config_t *config);
+
+
+/** Register a callback to monitor updates to the TLS configuration
+ *
+ * Register a callback function that will be invoked by the management thread whenever the sslProfile record associated
+ * with the qd_tls_config_t is updated.  Note that the update callback is invoked on the management thread while the
+ * qd_tls_config is locked. This prevents new TLS sessions from being created using the updated configuration until
+ * after the update callback returns.
+ *
+ * @param update_cb_context Opaque handle passed to update callback function.
+ * @param update_cb Optional callback when the sslProfile has been updated by management.
+ */
+typedef void (*qd_tls_config_update_cb_t)(const qd_tls_config_t *config,
+                                          void *update_cb_context);
+void qd_tls_config_register_update_callback(qd_tls_config_t *config, void *update_cb_context,
+                                            qd_tls_config_update_cb_t update_cb);
+
+
+/**
+ * Cancel the update callback.
+ *
+ * Deregisters the update callback provided in qd_tls_config(). No further calls to the callback will occur on return
+ * from this call. Can only be called from the context of the management thread.
+ *
+ * @param config The qd_tls_config_t whose callback will be cancelled
+ */
+void qd_tls_config_unregister_update_callback(qd_tls_config_t *config);
+
+
 /**
  * Release a TLS session context.
  *
@@ -153,6 +194,13 @@ char *qd_tls_session_get_protocol_ciphers(const qd_tls_session_t *session);
  */
 int qd_tls_session_get_ssf(const qd_tls_session_t *session);
 
+/**
+ * Get the ordinal of the sslProfile used to create this session
+ *
+ * @param session to be queried
+ * @return the value of the sslProfile ordinal associated with this session.
+ */
+uint64_t qd_tls_session_get_ssl_profile_ordinal(const qd_tls_session_t *session);
 
 /**
  * Fill out the given *profile with the configuration from the named sslProfile record.

python/skupper_router/management/skrouter.json

@@ -725,15 +725,15 @@
                     "description": "The absolute path to the file containing the unique id to display name mapping",
                     "create": true
                 },
-                "version": {
+                "ordinal": {
                     "type": "integer",
-                    "description": "RESERVED FOR FUTURE USE",
+                    "description": "Indicates the revision of the TLS certificates provided. Used by certificate rotation. Each time the TLS certificates associated with this profile are updated the ordinal value will be increased. The most recent version of the TLS credentials will have the highest numerical ordinal value.",
                     "create": true,
                     "update": true
                 },
-                "oldestValidVersion": {
+                "oldestValidOrdinal": {
                     "type": "integer",
-                    "description": "RESERVED FOR FUTURE USE",
+                    "description": "Used by certificate rotation to remove connections that were created using TLS certificates that are no longer considered valid. Any active connections based on TLS certificates an ordinal value less than oldesValidOrdinal will be immediately terminated by the router.",
                     "create": true,
                     "update": true
                 }

src/adaptors/amqp/amqp_adaptor.c

@@ -1376,53 +1376,39 @@ static void AMQP_opened_handler(qd_router_t *router, qd_connection_t *conn, bool
     char                   rversion[128];
     uint64_t               connection_id = qd_connection_connection_id(conn);
     pn_connection_t       *pn_conn = qd_connection_pn(conn);
-    pn_transport_t *tport = 0;
-    pn_sasl_t      *sasl  = 0;
-    const char     *mech  = 0;
-    const char     *user  = 0;
-    const char *container = conn->pn_conn ? pn_connection_remote_container(conn->pn_conn) : 0;
+    const char            *host   = 0;
+    uint64_t               group_ordinal = 0;
+    const char            *container  = conn->pn_conn ? pn_connection_remote_container(conn->pn_conn) : 0;
+    char                   group_correlator[QD_DISCRIMINATOR_SIZE];
+    char                   host_local[255];
+
+    rversion[0]         = 0;
+    group_correlator[0] = 0;
+    host_local[0]       = 0;
 
-    rversion[0] = 0;
     conn->strip_annotations_in  = false;
     conn->strip_annotations_out = false;
-    if (conn->pn_conn) {
-        tport = pn_connection_transport(conn->pn_conn);
-    }
-    if (tport) {
-        sasl = pn_sasl(tport);
-        if(conn->user_id)
-            user = conn->user_id;
-        else
-            user = pn_transport_get_user(tport);
-    }
-
-    if (sasl)
-        mech = pn_sasl_get_mech(sasl);
+    qd_router_connection_get_config(conn, &role, &cost, &name,
+                                    &conn->strip_annotations_in, &conn->strip_annotations_out, &link_capacity);
 
-    const char *host = 0;
-    char host_local[255];
-    const qd_server_config_t *config;
     qd_connector_t *connector = qd_connection_connector(conn);
-
     if (connector) {
-        config = qd_connector_get_config(connector);
+        const qd_server_config_t *config = qd_connector_get_server_config(connector);
         snprintf(host_local, 254, "%s", config->host_port);
         host = &host_local[0];
-    }
-    else
-        host = qd_connection_name(conn);
 
-
-    qd_router_connection_get_config(conn, &role, &cost, &name,
-                                    &conn->strip_annotations_in, &conn->strip_annotations_out, &link_capacity);
-
-    if (connector && !!connector->ctor_config->data_connection_count) {
-        memcpy(conn->group_correlator, connector->ctor_config->group_correlator, QD_DISCRIMINATOR_SIZE);
+        // Use the connectors tls_ordinal value as the group ordinal because the connection with the highest tls_ordinal
+        // value has the most up-to-date security credentials and should take precedence over connections with a lower
+        // ordinal value.
+        (void) qd_connector_get_tls_ordinal(connector, &group_ordinal);
+        memcpy(group_correlator, connector->ctor_config->group_correlator, QD_DISCRIMINATOR_SIZE);
         if (connector->is_data_connector) {
             // override the configured role to identify this as a data connection
             assert(role == QDR_ROLE_INTER_ROUTER);
             role = QDR_ROLE_INTER_ROUTER_DATA;
         }
+    } else {
+        host = qd_connection_name(conn);
     }
 
     // check offered capabilities for streaming link support and connection trunking support
@@ -1457,10 +1443,13 @@ static void AMQP_opened_handler(qd_router_t *router, qd_connection_t *conn, bool
         const bool is_router = (role == QDR_ROLE_INTER_ROUTER || role == QDR_ROLE_EDGE_CONNECTION);
         pn_data_rewind(props);
         if (pn_data_next(props) && pn_data_type(props) == PN_MAP) {
-            const size_t num_items = pn_data_get_map(props);
-            int props_found = 0;  // once all props found exit loop
+
+            const size_t num_items   = pn_data_get_map(props);
+            const int    max_props   = 8;  // total possible props
+            int          props_found = 0;  // once all props found exit loop
+
             pn_data_enter(props);
-            for (int i = 0; i < num_items / 2 && props_found < 7; ++i) {
+            for (int i = 0; i < num_items / 2 && props_found < max_props; ++i) {
                 if (!pn_data_next(props)) break;
                 if (pn_data_type(props) != PN_SYMBOL) break;  // invalid properties map
                 pn_bytes_t key = pn_data_get_symbol(props);
@@ -1493,11 +1482,26 @@ static void AMQP_opened_handler(qd_router_t *router, qd_connection_t *conn, bool
                 } else if (key.size == strlen(QD_CONNECTION_PROPERTY_GROUP_CORRELATOR_KEY) &&
                     strncmp(key.start, QD_CONNECTION_PROPERTY_GROUP_CORRELATOR_KEY, key.size) == 0) {
                     props_found += 1;
+                    assert(!connector);  // expect: connector sets correlator, listener consumes it
                     if (!pn_data_next(props)) break;
                     if (role == QDR_ROLE_INTER_ROUTER || role == QDR_ROLE_INTER_ROUTER_DATA) {
                         if (pn_data_type(props) == PN_STRING) {
+                            // pn_bytes is not null terminated
                             pn_bytes_t gc = pn_data_get_string(props);
-                            strncpy(conn->group_correlator, gc.start, MIN(gc.size, QD_DISCRIMINATOR_SIZE));
+                            size_t len = MIN(gc.size, QD_DISCRIMINATOR_BYTES);
+                            memcpy(group_correlator, gc.start, len);
+                            group_correlator[len] = '\0';
+                        }
+                    }
+
+                } else if (key.size == strlen(QD_CONNECTION_PROPERTY_GROUP_ORDINAL_KEY) &&
+                    strncmp(key.start, QD_CONNECTION_PROPERTY_GROUP_ORDINAL_KEY, key.size) == 0) {
+                    props_found += 1;
+                    assert(!connector);  // expect: connector sets ordinal, listener consumes it
+                    if (!pn_data_next(props)) break;
+                    if (role == QDR_ROLE_INTER_ROUTER || role == QDR_ROLE_INTER_ROUTER_DATA) {
+                        if (pn_data_type(props) == PN_ULONG) {
+                            group_ordinal = pn_data_get_ulong(props);
                         }
                     }
 
@@ -1530,6 +1534,7 @@ static void AMQP_opened_handler(qd_router_t *router, qd_connection_t *conn, bool
 
                 } else if ((key.size == strlen(QD_CONNECTION_PROPERTY_ACCESS_ID)
                            && strncmp(key.start, QD_CONNECTION_PROPERTY_ACCESS_ID, key.size) == 0)) {
+                    props_found += 1;
                     if (!pn_data_next(props)) break;
                     if (!!connector && !!connector->vflow_record && pn_data_type(props) == PN_STRING) {
                         vflow_set_ref_from_pn(connector->vflow_record, VFLOW_ATTRIBUTE_PEER, props);
@@ -1539,13 +1544,35 @@ static void AMQP_opened_handler(qd_router_t *router, qd_connection_t *conn, bool
                     // skip this key
                     if (!pn_data_next(props)) break;
                 }
+
+                // NOTE: if adding more keys update max_props value above!
             }
         }
     }
 
-    char *proto = 0;
-    char *cipher = 0;
-    int ssl_ssf = 0;
+    // Gather transport-level information
+
+    pn_transport_t *tport  = 0;
+    pn_sasl_t      *sasl   = 0;
+    const char     *mech   = 0;
+    const char     *user   = 0;
+    char           *proto  = 0;
+    char           *cipher = 0;
+    int            ssl_ssf = 0;
+
+    if (conn->pn_conn) {
+        tport = pn_connection_transport(conn->pn_conn);
+    }
+    if (tport) {
+        sasl = pn_sasl(tport);
+        if(conn->user_id)
+            user = conn->user_id;
+        else
+            user = pn_transport_get_user(tport);
+    }
+
+    if (sasl)
+        mech = pn_sasl_get_mech(sasl);
 
     if (conn->ssl) {
         proto = qd_tls_session_get_protocol_version(conn->ssl);
@@ -1567,13 +1594,14 @@ static void AMQP_opened_handler(qd_router_t *router, qd_connection_t *conn, bool
                                                                  (char*) user,
                                                                  container,
                                                                  props,
+                                                                 qd_tls_session_get_ssl_profile_ordinal(conn->ssl),
                                                                  ssl_ssf,
                                                                  !!conn->ssl,
                                                                  rversion,
                                                                  streaming_links,
                                                                  connection_trunking);
 
-    qdr_connection_info_set_group_correlator(connection_info, conn->group_correlator);
+    qdr_connection_info_set_group(connection_info, group_correlator, group_ordinal);
 
     qdr_connection_opened(router->router_core,
                           amqp_adaptor.adaptor,

src/adaptors/amqp/connection_manager.c

@@ -80,6 +80,8 @@ QD_EXPORT qd_listener_t *qd_dispatch_configure_listener(qd_dispatch_t *qd, qd_en
             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);
@@ -285,8 +287,8 @@ QD_EXPORT qd_error_t qd_entity_refresh_connector(qd_entity_t* entity, void *impl
 
 QD_EXPORT qd_connector_config_t *qd_dispatch_configure_connector(qd_dispatch_t *qd, qd_entity_t *entity)
 {
-    qd_connection_manager_t *cm       = qd->connection_manager;
-    qd_connector_config_t    *ctor_config = qd_connector_config_create(qd, entity);
+    qd_connection_manager_t *cm          = qd->connection_manager;
+    qd_connector_config_t   *ctor_config = qd_connector_config_create(qd, entity);
     if (!ctor_config) {
         return 0;
     }