tools: Add tool to connect IPCP components

This enables user-written tools to instruct IPCPs to establish and
tear down connections (a.k.a. adjacencies) between its internal
components (Management and Data Transfer).

For more info, do "irm ipcp connect" or "irm ipcp disconnect" on the
command line.

This commit exposes a deletion bug in the RIB where FSO's fail to
unpack/parse. This will be fixed when the RIB is deprecated.

12 files changed, 178 insertions, 33 deletions

diff --git a/src/ipcpd/ipcp.c b/src/ipcpd/ipcp.c
index d4b3a7fc..41ea4784 100644
--- a/src/ipcpd/ipcp.c
+++ b/src/ipcpd/ipcp.c
@@ -301,6 +301,32 @@ static void * mainloop(void * o)
                                 dif_info.dif_name      = info.dif_name;
                         }
                         break;
+                case IPCP_MSG_CODE__IPCP_CONNECT:
+                        ret_msg.has_result = true;
+
+                        if (ipcpi.ops->ipcp_connect == NULL) {
+                                log_err("Connect unsupported.");
+                                ret_msg.result = -ENOTSUP;
+                                break;
+                        }
+
+                        ret_msg.result =
+                                ipcpi.ops->ipcp_connect(msg->dst_name,
+                                                        msg->comp_name);
+                        break;
+                case IPCP_MSG_CODE__IPCP_DISCONNECT:
+                        ret_msg.has_result = true;
+
+                        if (ipcpi.ops->ipcp_disconnect == NULL) {
+                                log_err("Disconnect unsupported.");
+                                ret_msg.result = -ENOTSUP;
+                                break;
+                        }
+
+                        ret_msg.result =
+                                ipcpi.ops->ipcp_disconnect(msg->dst_name,
+                                                           msg->comp_name);
+                        break;
                 case IPCP_MSG_CODE__IPCP_REG:
                         ret_msg.has_result = true;
 
@@ -435,6 +461,8 @@ static void * mainloop(void * o)
                                 ipcpi.ops->ipcp_flow_dealloc(fd);
                         break;
                 default:
+                        ret_msg.has_result = true;
+                        ret_msg.result     = -1;
                         log_err("Don't know that message code");
                         break;
                 }
diff --git a/src/ipcpd/ipcp.h b/src/ipcpd/ipcp.h
index cd18d198..1b2a0334 100644
--- a/src/ipcpd/ipcp.h
+++ b/src/ipcpd/ipcp.h
@@ -45,6 +45,12 @@ struct ipcp_ops {
         int   (* ipcp_enroll)(const char *      dst,
                               struct dif_info * info);
 
+        int   (* ipcp_connect)(const char * dst,
+                               const char * component);
+
+        int   (* ipcp_disconnect)(const char * dst,
+                                  const char * component);
+
         int   (* ipcp_reg)(const uint8_t * hash);
 
         int   (* ipcp_unreg)(const uint8_t * hash);
diff --git a/src/ipcpd/local/main.c b/src/ipcpd/local/main.c
index 37d23fc3..c6f88d78 100644
--- a/src/ipcpd/local/main.c
+++ b/src/ipcpd/local/main.c
@@ -323,7 +323,9 @@ static int ipcp_local_flow_dealloc(int fd)
 
 static struct ipcp_ops local_ops = {
         .ipcp_bootstrap       = ipcp_local_bootstrap,
-        .ipcp_enroll          = NULL,                       /* shim */
+        .ipcp_enroll          = NULL,
+        .ipcp_connect         = NULL,
+        .ipcp_disconnect      = NULL,
         .ipcp_reg             = ipcp_local_reg,
         .ipcp_unreg           = ipcp_local_unreg,
         .ipcp_query           = ipcp_local_query,
diff --git a/src/ipcpd/normal/ae.h b/src/ipcpd/normal/ae.h
index 4534cefa..3d3bdc27 100644
--- a/src/ipcpd/normal/ae.h
+++ b/src/ipcpd/normal/ae.h
@@ -27,6 +27,8 @@
 
 #include "dt.h"
 
+#define DST_MAX_STRLEN 64
+
 enum ae_id {
         AEID_DT = 0,
         AEID_ENROLL,
@@ -36,9 +38,10 @@ enum ae_id {
 
 struct conn {
         struct conn_info conn_info;
-        struct flow_info {
-                int              fd;
-                qosspec_t        qs;
+        struct {
+                char      dst[DST_MAX_STRLEN + 1];
+                int       fd;
+                qosspec_t qs;
         } flow_info;
 };
 
diff --git a/src/ipcpd/normal/connmgr.c b/src/ipcpd/normal/connmgr.c
index b6e5e31a..11f83247 100644
--- a/src/ipcpd/normal/connmgr.c
+++ b/src/ipcpd/normal/connmgr.c
@@ -48,7 +48,7 @@ enum connmgr_state {
         CONNMGR_RUNNING
 };
 
-struct ae_conn {
+struct conn_el {
         struct list_head next;
         struct conn      conn;
 };
@@ -58,15 +58,17 @@ struct ae {
         struct conn_info info;
 
         struct list_head conns;
+        struct list_head pending;
+
         pthread_cond_t   cond;
         pthread_mutex_t  lock;
 };
 
 struct {
-        struct ae                aes[AEID_MAX];
-        enum connmgr_state       state;
+        struct ae          aes[AEID_MAX];
+        enum connmgr_state state;
 
-        pthread_t                acceptor;
+        pthread_t          acceptor;
 } connmgr;
 
 static int get_id_by_name(const char * name)
@@ -85,23 +87,21 @@ static int add_ae_conn(enum ae_id         id,
                        qosspec_t          qs,
                        struct conn_info * rcv_info)
 {
-        struct ae_conn *   ae_conn;
+        struct conn_el * el;
 
-        ae_conn = malloc(sizeof(*ae_conn));
-        if (ae_conn == NULL) {
+        el = malloc(sizeof(*el));
+        if (el == NULL) {
                 log_err("Not enough memory.");
                 return -1;
         }
 
-        ae_conn->conn.conn_info    = *rcv_info;
-        ae_conn->conn.flow_info.fd = fd;
-        ae_conn->conn.flow_info.qs = qs;
-
-        list_head_init(&ae_conn->next);
+        el->conn.conn_info    = *rcv_info;
+        el->conn.flow_info.fd = fd;
+        el->conn.flow_info.qs = qs;
 
         pthread_mutex_lock(&connmgr.aes[id].lock);
 
-        list_add(&ae_conn->next, &connmgr.aes[id].conns);
+        list_add(&el->next, &connmgr.aes[id].pending);
         pthread_cond_signal(&connmgr.aes[id].cond);
 
         pthread_mutex_unlock(&connmgr.aes[id].lock);
@@ -217,6 +217,7 @@ int connmgr_ae_init(enum ae_id               id,
         }
 
         list_head_init(&ae->conns);
+        list_head_init(&ae->pending);
 
         memcpy(&connmgr.aes[id].info, info, sizeof(connmgr.aes[id].info));
 
@@ -241,7 +242,13 @@ void connmgr_ae_fini(enum ae_id id)
         pthread_mutex_lock(&ae->lock);
 
         list_for_each_safe(p, h, &ae->conns) {
-                struct ae_conn * e = list_entry(p, struct ae_conn, next);
+                struct conn_el * e = list_entry(p, struct conn_el, next);
+                list_del(&e->next);
+                free(e);
+        }
+
+        list_for_each_safe(p, h, &ae->pending) {
+                struct conn_el * e = list_entry(p, struct conn_el, next);
                 list_del(&e->next);
                 free(e);
         }
@@ -256,6 +263,84 @@ void connmgr_ae_fini(enum ae_id id)
         connmgr.aes[id].nbs = NULL;
 }
 
+int connmgr_ipcp_connect(const char * dst,
+                         const char * component)
+{
+        struct conn_el * ce;
+        int              id;
+
+        assert(dst);
+        assert(component);
+
+        ce = malloc(sizeof(*ce));
+        if (ce == NULL) {
+                log_dbg("Out of memory.");
+                return -1;
+        }
+
+        id = get_id_by_name(component);
+        if (id < 0) {
+                log_dbg("No such component: %s", component);
+                free(ce);
+                return -1;
+        }
+
+        /* FIXME: get the correct qos for the component. */
+        if (connmgr_alloc(id, dst, NULL, &ce->conn)) {
+                free(ce);
+                return -1;
+        }
+
+        if (strlen(dst) > DST_MAX_STRLEN) {
+                log_warn("Truncating dst length for connection.");
+                memcpy(ce->conn.flow_info.dst, dst, DST_MAX_STRLEN);
+                ce->conn.flow_info.dst[DST_MAX_STRLEN] = '\0';
+        } else {
+                strcpy(ce->conn.flow_info.dst, dst);
+        }
+
+        pthread_mutex_lock(&connmgr.aes[id].lock);
+
+        list_add(&ce->next, &connmgr.aes[id].conns);
+
+        pthread_mutex_unlock(&connmgr.aes[id].lock);
+
+        return 0;
+}
+
+int connmgr_ipcp_disconnect(const char * dst,
+                            const char * component)
+{
+        struct list_head * p;
+        struct list_head * h;
+        int                id;
+
+        assert(dst);
+        assert(component);
+
+        id = get_id_by_name(component);
+        if (id < 0)
+                return -1;
+
+        pthread_mutex_lock(&connmgr.aes[id].lock);
+
+        list_for_each_safe(p,h, &connmgr.aes[id].conns) {
+                struct conn_el * el = list_entry(p, struct conn_el, next);
+                if (strcmp(el->conn.flow_info.dst, dst) == 0) {
+                        int ret;
+                        pthread_mutex_unlock(&connmgr.aes[id].lock);
+                        list_del(&el->next);
+                        ret = connmgr_dealloc(id, &el->conn);
+                        free(el);
+                        return ret;
+                }
+        }
+
+        pthread_mutex_unlock(&connmgr.aes[id].lock);
+
+        return 0;
+}
+
 int connmgr_alloc(enum ae_id    id,
                   const char *  dst,
                   qosspec_t *   qs,
@@ -329,7 +414,7 @@ int connmgr_dealloc(enum ae_id    id,
 int connmgr_wait(enum ae_id    id,
                  struct conn * conn)
 {
-        struct ae_conn * ae_conn;
+        struct conn_el * el;
         struct ae *      ae;
 
         assert(id >= 0 && id < AEID_MAX);
@@ -342,21 +427,21 @@ int connmgr_wait(enum ae_id    id,
         pthread_cleanup_push((void(*)(void *))pthread_mutex_unlock,
                              (void *) &ae->lock);
 
-        while (list_is_empty(&ae->conns))
+        while (list_is_empty(&ae->pending))
                 pthread_cond_wait(&ae->cond, &ae->lock);
 
         pthread_cleanup_pop(false);
 
-        ae_conn = list_first_entry((&ae->conns), struct ae_conn, next);
-        if (ae_conn == NULL) {
+        el = list_first_entry((&ae->pending), struct conn_el, next);
+        if (el == NULL) {
                 pthread_mutex_unlock(&ae->lock);
                 return -1;
         }
 
-        *conn = ae_conn->conn;
+        *conn = el->conn;
 
-        list_del(&ae_conn->next);
-        free(ae_conn);
+        list_del(&el->next);
+        free(el);
 
         pthread_mutex_unlock(&ae->lock);
 
diff --git a/src/ipcpd/normal/connmgr.h b/src/ipcpd/normal/connmgr.h
index 920058da..379877e6 100644
--- a/src/ipcpd/normal/connmgr.h
+++ b/src/ipcpd/normal/connmgr.h
@@ -43,6 +43,12 @@ int         connmgr_ae_init(enum ae_id               id,
 
 void        connmgr_ae_fini(enum ae_id id);
 
+int         connmgr_ipcp_connect(const char * dst,
+                                 const char * component);
+
+int         connmgr_ipcp_disconnect(const char * dst,
+                                    const char * component);
+
 int         connmgr_alloc(enum ae_id    id,
                           const char *  dst,
                           qosspec_t *   qs,
diff --git a/src/ipcpd/normal/dt.c b/src/ipcpd/normal/dt.c
index 19c2d3a6..aa089852 100644
--- a/src/ipcpd/normal/dt.c
+++ b/src/ipcpd/normal/dt.c
@@ -67,7 +67,6 @@ struct {
         pthread_rwlock_t   lock;
 
         struct nbs *       nbs;
-        struct ae *        ae;
 
         struct nb_notifier nb_notifier;
 
@@ -163,6 +162,8 @@ static void * dt_conn_handle(void * o)
 
                 log_dbg("Got new connection.");
 
+                /* NOTE: connection acceptance policy could be here. */
+
                 nbs_add(dt.nbs, conn);
         }
 
diff --git a/src/ipcpd/normal/enroll.c b/src/ipcpd/normal/enroll.c
index 2f7dd9bc..7f93ed3a 100644
--- a/src/ipcpd/normal/enroll.c
+++ b/src/ipcpd/normal/enroll.c
@@ -259,14 +259,17 @@ static void * enroll_handle(void * o)
                         continue;
                 }
 
-                if (msg->code != ENROLL_CODE__ENROLL_DONE) {
+                if (msg->code != ENROLL_CODE__ENROLL_DONE || !msg->has_result) {
                         log_err("Wrong message type.");
                         enroll_msg__free_unpacked(msg, NULL);
                         connmgr_dealloc(AEID_ENROLL, &conn);
                         continue;
                 }
 
-                log_dbg("Neighbor enrollment successful.");
+                if (msg->result == 0)
+                        log_dbg("Neighbor enrollment successful.");
+                else
+                        log_dbg("Neigbor reported failed enrollment.");
 
                 connmgr_dealloc(AEID_ENROLL, &conn);
         }
@@ -287,13 +290,16 @@ int enroll_boot(struct conn * conn,
         return 0;
 }
 
-int enroll_done(struct conn * conn)
+int enroll_done(struct conn * conn,
+                int           result)
 {
         enroll_msg_t msg = ENROLL_MSG__INIT;
         uint8_t      buf[ENROLL_BUF_LEN];
         ssize_t       len;
 
-        msg.code = ENROLL_CODE__ENROLL_DONE;
+        msg.code       = ENROLL_CODE__ENROLL_DONE;
+        msg.has_result = true;
+        msg.result     = result;
 
         len = enroll_msg__get_packed_size(&msg);
         if (len < 0) {
diff --git a/src/ipcpd/normal/enroll.h b/src/ipcpd/normal/enroll.h
index 67c9912d..3b277e44 100644
--- a/src/ipcpd/normal/enroll.h
+++ b/src/ipcpd/normal/enroll.h
@@ -40,7 +40,8 @@ void                 enroll_bootstrap(const struct ipcp_config * conf);
 int                  enroll_boot(struct conn * conn,
                                  const char *  dst);
 
-int                  enroll_done(struct conn * conn);
+int                  enroll_done(struct conn * conn,
+                                 int           result);
 
 struct ipcp_config * enroll_get_conf(void);
 
diff --git a/src/ipcpd/normal/main.c b/src/ipcpd/normal/main.c
index d9fbc2dd..bef04b7a 100644
--- a/src/ipcpd/normal/main.c
+++ b/src/ipcpd/normal/main.c
@@ -263,11 +263,12 @@ static int normal_ipcp_enroll(const char *      dst,
         }
 
         if (enroll_components(dt_conn.conn_info.addr)) {
+                enroll_done(&er_conn, -1);
                 log_err("Failed to enroll components.");
                 goto fail_enroll_comp;
         }
 
-        if (enroll_done(&er_conn))
+        if (enroll_done(&er_conn, 0))
                 log_warn("Failed to confirm enrollment with peer.");
 
         if (connmgr_dealloc(AEID_DT, &dt_conn))
@@ -346,6 +347,8 @@ static int normal_ipcp_query(const uint8_t * dst)
 static struct ipcp_ops normal_ops = {
         .ipcp_bootstrap       = normal_ipcp_bootstrap,
         .ipcp_enroll          = normal_ipcp_enroll,
+        .ipcp_connect         = connmgr_ipcp_connect,
+        .ipcp_disconnect      = connmgr_ipcp_disconnect,
         .ipcp_reg             = dir_reg,
         .ipcp_unreg           = dir_unreg,
         .ipcp_query           = normal_ipcp_query,
diff --git a/src/ipcpd/shim-eth-llc/main.c b/src/ipcpd/shim-eth-llc/main.c
index bcf5abe2..3d186d7a 100644
--- a/src/ipcpd/shim-eth-llc/main.c
+++ b/src/ipcpd/shim-eth-llc/main.c
@@ -1216,6 +1216,8 @@ static int eth_llc_ipcp_flow_dealloc(int fd)
 static struct ipcp_ops eth_llc_ops = {
         .ipcp_bootstrap       = eth_llc_ipcp_bootstrap,
         .ipcp_enroll          = NULL,
+        .ipcp_connect         = NULL,
+        .ipcp_disconnect      = NULL,
         .ipcp_reg             = eth_llc_ipcp_reg,
         .ipcp_unreg           = eth_llc_ipcp_unreg,
         .ipcp_query           = eth_llc_ipcp_query,
diff --git a/src/ipcpd/shim-udp/main.c b/src/ipcpd/shim-udp/main.c
index 55fe19a6..0bf57741 100644
--- a/src/ipcpd/shim-udp/main.c
+++ b/src/ipcpd/shim-udp/main.c
@@ -1092,7 +1092,9 @@ static int ipcp_udp_flow_dealloc(int fd)
 
 static struct ipcp_ops udp_ops = {
         .ipcp_bootstrap       = ipcp_udp_bootstrap,
-        .ipcp_enroll          = NULL,                       /* shim */
+        .ipcp_enroll          = NULL,
+        .ipcp_connect         = NULL,
+        .ipcp_disconnect      = NULL,
         .ipcp_reg             = ipcp_udp_reg,
         .ipcp_unreg           = ipcp_udp_unreg,
         .ipcp_query           = ipcp_udp_query,