ipcpd: Add flow-update relay

This adds an ipcp_flow_update() call to relay opaque messages between
the two IRMds (carried by FLOW_IRM_UPDATE messages), which passes it
back up to the peer IRMd via ipcp_flow_update_arr(). The broadcast
layer does not implement this.

Needed for periodic re-keying of encrypted flows via OAP.

Signed-off-by: Dimitri Staessens <dimitri@ouroboros.rocks>
Signed-off-by: Sander Vrijders <sander@ouroboros.rocks>

17 files changed, 503 insertions, 7 deletions

diff --git a/include/ouroboros/ipcp-dev.h b/include/ouroboros/ipcp-dev.h
index d23f757e..d00d6f08 100644
--- a/include/ouroboros/ipcp-dev.h
+++ b/include/ouroboros/ipcp-dev.h
@@ -38,6 +38,9 @@ int    ipcp_flow_req_arr(const buffer_t * dst,
                          uint32_t         mtu,
                          const buffer_t * data);
 
+int    ipcp_flow_update_arr(int              flow_id,
+                            const buffer_t * data);
+
 int    ipcp_flow_alloc_reply(int              fd,
                              int              response,
                              time_t           mpl,
diff --git a/include/ouroboros/np1_flow.h b/include/ouroboros/np1_flow.h
index 309d01c2..758b6db8 100644
--- a/include/ouroboros/np1_flow.h
+++ b/include/ouroboros/np1_flow.h
@@ -36,6 +36,10 @@ int  np1_flow_resp(int flow_id,
 int  np1_flow_dealloc(int    flow_id,
                       time_t timeo);
 
+int  np1_flow_fd(int flow_id);
+
+int  np1_flow_id(int fd);
+
 static const qosspec_t qos_np1 = {
         .service      = SVC_RAW,
         .delay        = UINT32_MAX,
diff --git a/include/ouroboros/serdes-irm.h b/include/ouroboros/serdes-irm.h
index 1dfff4d9..01c4153c 100644
--- a/include/ouroboros/serdes-irm.h
+++ b/include/ouroboros/serdes-irm.h
@@ -31,6 +31,7 @@
 #include <ouroboros/utils.h>
 
 #include <inttypes.h>
+#include <stdbool.h>
 
 int flow_alloc__irm_req_ser(buffer_t *               buf,
                             const struct flow_info * flow,
@@ -51,6 +52,10 @@ int ipcp_flow_req_arr__irm_req_ser(buffer_t *               buf,
                                    const struct flow_info * flow,
                                    const buffer_t *         data);
 
+int ipcp_flow_update_arr__irm_req_ser(buffer_t *               buf,
+                                      const struct flow_info * flow,
+                                      const buffer_t *         data);
+
 int ipcp_flow_alloc_reply__irm_msg_ser(buffer_t *               buf,
                                        const struct flow_info * flow,
                                        int                      response,
@@ -64,6 +69,14 @@ int flow_dealloc__irm_req_ser(buffer_t *               buf,
                               const struct flow_info * flow,
                               const struct timespec *  timeo);
 
+int flow_update__irm_req_ser(buffer_t *               buf,
+                             const struct flow_info * flow,
+                             bool                     rekey);
+
+int flow_rekey__irm_result_des(buffer_t *        buf,
+                               struct crypt_sk * sk,
+                               bool *            has_key);
+
 int ipcp_flow_dealloc__irm_req_ser(buffer_t *               buf,
                                    const struct flow_info * info);
 
diff --git a/src/ipcpd/eth/eth.c b/src/ipcpd/eth/eth.c
index d6f476f2..7e038a03 100644
--- a/src/ipcpd/eth/eth.c
+++ b/src/ipcpd/eth/eth.c
@@ -49,6 +49,7 @@
 #include <ouroboros/ipcp-dev.h>
 #include <ouroboros/fqueue.h>
 #include <ouroboros/logs.h>
+#include <ouroboros/np1_flow.h>
 #include <ouroboros/time.h>
 #include <ouroboros/fccntl.h>
 #include <ouroboros/pthread.h>
@@ -160,6 +161,7 @@
 #define FLOW_REPLY           1
 #define NAME_QUERY_REQ       2
 #define NAME_QUERY_REPLY     3
+#define FLOW_IRM_UPDATE      4
 
 struct mgmt_msg {
 #if defined(BUILD_ETH_DIX)
@@ -844,6 +846,58 @@ static int eth_ipcp_alloc_resp(uint8_t *        dst_addr,
         return 0;
 }
 
+static int eth_ipcp_flow_update(int              fd,
+                                const buffer_t * data)
+{
+        struct mgmt_msg * msg;
+        struct ef *       flow;
+        uint8_t *         buf;
+        uint8_t           r_addr[MAC_SIZE];
+        int               ret;
+
+        buf = malloc(sizeof(*msg) + ETH_HEADER_TOT_SIZE + data->len);
+        if (buf == NULL)
+                return -1;
+
+        memset(buf, 0, sizeof(*msg) + ETH_HEADER_TOT_SIZE + data->len);
+
+        msg = (struct mgmt_msg *) (buf + ETH_HEADER_TOT_SIZE);
+
+        msg->code = FLOW_IRM_UPDATE;
+
+        pthread_rwlock_rdlock(&eth_data.flows_lock);
+
+        flow = &eth_data.fd_to_ef[fd];
+#if defined(BUILD_ETH_DIX)
+        msg->seid = htons((uint16_t) fd);
+        msg->deid = htons((uint16_t) flow->r_eid);
+#elif defined(BUILD_ETH_LLC)
+        msg->ssap = flow->sap;
+        msg->dsap = (uint8_t) flow->r_sap;
+#endif
+        memcpy(r_addr, flow->r_addr, MAC_SIZE);
+
+        pthread_rwlock_unlock(&eth_data.flows_lock);
+
+        if (data->len > 0)
+                memcpy(msg + 1, data->data, data->len);
+
+        ret = eth_ipcp_send_frame(r_addr,
+#if defined(BUILD_ETH_DIX)
+                                  MGMT_EID,
+#elif defined(BUILD_ETH_LLC)
+                                  reverse_bits(MGMT_SAP),
+                                  reverse_bits(MGMT_SAP),
+#endif
+                                  buf, sizeof(*msg) + data->len);
+        free(buf);
+
+        if (ret == 0)
+                FETCH_ADD_RELAXED(&eth_data.stat.n_mgmt_snd, 1);
+
+        return ret;
+}
+
 static int eth_ipcp_req(uint8_t *        r_addr,
 #if defined(BUILD_ETH_DIX)
                         uint16_t         r_eid,
@@ -994,6 +1048,44 @@ static int eth_ipcp_name_query_reply(const uint8_t * hash,
         return 0;
 }
 
+static int eth_ipcp_flow_update_arr(const uint8_t * buf,
+                                    size_t          len)
+{
+        struct mgmt_msg * msg;
+        buffer_t          data;
+        int               fd;
+        int               flow_id;
+
+        msg = (struct mgmt_msg *) buf;
+
+        data.data = (uint8_t *) buf + sizeof(*msg);
+        data.len  = len - sizeof(*msg);
+
+        pthread_rwlock_rdlock(&eth_data.flows_lock);
+#if defined(BUILD_ETH_DIX)
+        fd = ntohs(msg->deid);
+#elif defined(BUILD_ETH_LLC)
+        fd = eth_data.ef_to_fd[msg->dsap];
+#endif
+        pthread_rwlock_unlock(&eth_data.flows_lock);
+
+        if (fd < 0 || fd >= SYS_MAX_FLOWS) {
+                log_err("Flow update for unknown endpoint.");
+                return -1;
+        }
+
+        flow_id = np1_flow_id(fd);
+        if (flow_id < 0)
+                return -1;
+
+        if (ipcp_flow_update_arr(flow_id, &data) < 0) {
+                log_err("Failed to relay flow update on fd %d.", fd);
+                return -1;
+        }
+
+        return 0;
+}
+
 static int eth_ipcp_mgmt_frame(const uint8_t * buf,
                                size_t          len,
                                uint8_t *       r_addr)
@@ -1055,6 +1147,9 @@ static int eth_ipcp_mgmt_frame(const uint8_t * buf,
                                      ntoh32(msg->response),
                                      &data);
                 break;
+        case FLOW_IRM_UPDATE:
+                eth_ipcp_flow_update_arr(buf, len);
+                break;
         case NAME_QUERY_REQ:
                 if (len < sizeof(*msg) + ipcp_dir_hash_len())
                         return -1;
@@ -2289,7 +2384,8 @@ static struct ipcp_ops eth_ops = {
         .ipcp_flow_alloc      = eth_ipcp_flow_alloc,
         .ipcp_flow_join       = NULL,
         .ipcp_flow_alloc_resp = eth_ipcp_flow_alloc_resp,
-        .ipcp_flow_dealloc    = eth_ipcp_flow_dealloc
+        .ipcp_flow_dealloc    = eth_ipcp_flow_dealloc,
+        .ipcp_flow_update     = eth_ipcp_flow_update
 };
 
 int main(int    argc,
diff --git a/src/ipcpd/ipcp.c b/src/ipcpd/ipcp.c
index 1052a686..dcee4b9c 100644
--- a/src/ipcpd/ipcp.c
+++ b/src/ipcpd/ipcp.c
@@ -820,6 +820,33 @@ static void do_flow_dealloc(int          flow_id,
         log_info("Finished deallocating flow %d.", flow_id);
 }
 
+static void do_flow_update(int              flow_id,
+                           const buffer_t * data,
+                           ipcp_msg_t *     ret_msg)
+{
+        int fd;
+
+        if (ipcpd.ops->ipcp_flow_update == NULL) {
+                log_err("Failed to update flow: operation unsupported.");
+                ret_msg->result = -ENOTSUP;
+                return;
+        }
+
+        if (ipcp_get_state() != IPCP_OPERATIONAL) {
+                ret_msg->result = -EIPCPSTATE;
+                return;
+        }
+
+        fd = np1_flow_fd(flow_id);
+        if (fd < 0) {
+                log_warn("Flow update for unknown flow_id %d.", flow_id);
+                ret_msg->result = -1;
+                return;
+        }
+
+        ret_msg->result = ipcpd.ops->ipcp_flow_update(fd, data);
+}
+
 static void * mainloop(void * o)
 {
         int          sfd;
@@ -918,6 +945,13 @@ static void * mainloop(void * o)
                 case IPCP_MSG_CODE__IPCP_FLOW_DEALLOC:
                         do_flow_dealloc(msg->flow_id, msg->timeo_sec, &ret_msg);
                         break;
+                case IPCP_MSG_CODE__IPCP_FLOW_UPDATE:
+                        assert(msg->pk.len > 0 ? msg->pk.data != NULL
+                                               : msg->pk.data == NULL);
+                        data.len = msg->pk.len;
+                        data.data = msg->pk.data;
+                        do_flow_update(msg->flow_id, &data, &ret_msg);
+                        break;
                 default:
                         ret_msg.result = -1;
                         log_err("Unknown message code: %d.", msg->code);
diff --git a/src/ipcpd/ipcp.h b/src/ipcpd/ipcp.h
index 0adcc694..210157ec 100644
--- a/src/ipcpd/ipcp.h
+++ b/src/ipcpd/ipcp.h
@@ -68,6 +68,9 @@ struct ipcp_ops {
                                        const buffer_t * data);
 
         int   (* ipcp_flow_dealloc)(int fd);
+
+        int   (* ipcp_flow_update)(int              fd,
+                                   const buffer_t * data);
 };
 
 int             ipcp_init(int               argc,
diff --git a/src/ipcpd/local/main.c b/src/ipcpd/local/main.c
index eb9836f2..c0aeb51e 100644
--- a/src/ipcpd/local/main.c
+++ b/src/ipcpd/local/main.c
@@ -38,6 +38,7 @@
 #include <ouroboros/ipcp.h>
 #include <ouroboros/ipcp-dev.h>
 #include <ouroboros/local-dev.h>
+#include <ouroboros/np1_flow.h>
 
 #include "ipcp.h"
 #include "np1.h"
@@ -297,6 +298,38 @@ static int local_ipcp_flow_dealloc(int fd)
         return 0;
 }
 
+/* Loopback relay: deliver the update back to the peer end (same IRMd). */
+static int local_ipcp_flow_update(int              fd,
+                                  const buffer_t * data)
+{
+        int out_fd;
+        int out_flow_id;
+
+        pthread_rwlock_rdlock(&local_data.lock);
+
+        out_fd = local_data.in_out[fd];
+
+        pthread_rwlock_unlock(&local_data.lock);
+
+        if (out_fd == -1) {
+                log_err("Flow update on fd %d with no peer.", fd);
+                return -1;
+        }
+
+        out_flow_id = np1_flow_id(out_fd);
+        if (out_flow_id < 0) {
+                log_err("No flow_id for peer fd %d.", out_fd);
+                return -1;
+        }
+
+        if (ipcp_flow_update_arr(out_flow_id, data) < 0) {
+                log_err("Failed to relay flow update to fd %d.", out_fd);
+                return -1;
+        }
+
+        return 0;
+}
+
 static struct ipcp_ops local_ops = {
         .ipcp_bootstrap       = local_ipcp_bootstrap,
         .ipcp_enroll          = NULL,
@@ -308,7 +341,8 @@ static struct ipcp_ops local_ops = {
         .ipcp_flow_alloc      = local_ipcp_flow_alloc,
         .ipcp_flow_join       = NULL,
         .ipcp_flow_alloc_resp = local_ipcp_flow_alloc_resp,
-        .ipcp_flow_dealloc    = local_ipcp_flow_dealloc
+        .ipcp_flow_dealloc    = local_ipcp_flow_dealloc,
+        .ipcp_flow_update     = local_ipcp_flow_update
 };
 
 int main(int    argc,
diff --git a/src/ipcpd/udp/udp.c b/src/ipcpd/udp/udp.c
index 93e88b9b..db57e2f4 100644
--- a/src/ipcpd/udp/udp.c
+++ b/src/ipcpd/udp/udp.c
@@ -28,6 +28,8 @@
 #include <ouroboros/list.h>
 #include <ouroboros/utils.h>
 #include <ouroboros/dev.h>
+#include <ouroboros/ipcp-dev.h>
+#include <ouroboros/np1_flow.h>
 #include <ouroboros/fqueue.h>
 #include <ouroboros/errno.h>
 #include <ouroboros/logs.h>
@@ -54,6 +56,7 @@
 
 #define FLOW_REQ                 1
 #define FLOW_REPLY               2
+#define FLOW_IRM_UPDATE          3
 
 #define OUR_HEADER_LEN           sizeof(uint32_t) /* adds eid */
 
@@ -328,6 +331,48 @@ static int udp_ipcp_port_alloc_resp(const struct __SOCKADDR * r_saddr,
         return 0;
 }
 
+static int udp_ipcp_flow_update(int              fd,
+                                const buffer_t * data)
+{
+        struct mgmt_msg * msg;
+        struct __SOCKADDR r_saddr;
+        uint32_t          d_eid;
+
+        msg = malloc(sizeof(*msg) + data->len);
+        if (msg == NULL)
+                return -1;
+
+        memset(msg, 0, sizeof(*msg) + data->len);
+
+        pthread_rwlock_rdlock(&udp_data.flows_lock);
+
+        r_saddr = udp_data.fd_to_uf[fd].r_saddr;
+        d_eid   = (uint32_t) udp_data.fd_to_uf[fd].d_eid;
+
+        pthread_rwlock_unlock(&udp_data.flows_lock);
+
+        msg->eid   = hton32(MGMT_EID);
+        msg->code  = FLOW_IRM_UPDATE;
+        msg->s_eid = hton32(d_eid);
+        msg->d_eid = hton32((uint32_t) fd);
+
+        if (data->len > 0)
+                memcpy(msg + 1, data->data, data->len);
+
+        if (sendto(udp_data.s_fd, msg, sizeof(*msg) + data->len,
+                   SENDTO_FLAGS,
+                   (const struct sockaddr *) &r_saddr,
+                   sizeof(r_saddr)) < 0) {
+                log_err("Failed to send flow update: %s.", strerror(errno));
+                free(msg);
+                return -1;
+        }
+
+        free(msg);
+
+        return 0;
+}
+
 static int udp_ipcp_port_req(struct __SOCKADDR * c_saddr,
                              int                 d_eid,
                              const uint8_t *     dst,
@@ -396,6 +441,37 @@ static int udp_ipcp_port_alloc_reply(const struct __SOCKADDR * saddr,
         return 0;
 }
 
+static int udp_ipcp_flow_update_arr(const uint8_t * buf,
+                                    size_t          len)
+{
+        struct mgmt_msg * msg;
+        buffer_t          data;
+        int               fd;
+        int               flow_id;
+
+        msg = (struct mgmt_msg *) buf;
+
+        fd = (int) ntoh32(msg->s_eid);
+        if (fd < 0 || fd >= SYS_MAX_FLOWS) {
+                log_err("Flow update for invalid eid %d.", fd);
+                return -1;
+        }
+
+        data.len  = len - sizeof(*msg);
+        data.data = (uint8_t *) buf + sizeof(*msg);
+
+        flow_id = np1_flow_id(fd);
+        if (flow_id < 0)
+                return -1;
+
+        if (ipcp_flow_update_arr(flow_id, &data) < 0) {
+                log_err("Failed to relay flow update on fd %d.", fd);
+                return -1;
+        }
+
+        return 0;
+}
+
 static int udp_ipcp_mgmt_frame(struct __SOCKADDR c_saddr,
                                const uint8_t *   buf,
                                size_t            len)
@@ -443,6 +519,8 @@ static int udp_ipcp_mgmt_frame(struct __SOCKADDR c_saddr,
                                                  ntoh32(msg->d_eid),
                                                  ntoh32(msg->response),
                                                  &data);
+        case FLOW_IRM_UPDATE:
+                return udp_ipcp_flow_update_arr(buf, len);
         default:
                 log_err("Unknown message received %d.", msg->code);
                 return -1;
@@ -1196,7 +1274,8 @@ static struct ipcp_ops udp_ops = {
         .ipcp_flow_alloc      = udp_ipcp_flow_alloc,
         .ipcp_flow_join       = NULL,
         .ipcp_flow_alloc_resp = udp_ipcp_flow_alloc_resp,
-        .ipcp_flow_dealloc    = udp_ipcp_flow_dealloc
+        .ipcp_flow_dealloc    = udp_ipcp_flow_dealloc,
+        .ipcp_flow_update     = udp_ipcp_flow_update
 };
 
 int main(int    argc,
diff --git a/src/ipcpd/unicast/fa.c b/src/ipcpd/unicast/fa.c
index 43c56f90..c6eca175 100644
--- a/src/ipcpd/unicast/fa.c
+++ b/src/ipcpd/unicast/fa.c
@@ -37,6 +37,7 @@
 #include <ouroboros/errno.h>
 #include <ouroboros/dev.h>
 #include <ouroboros/ipcp-dev.h>
+#include <ouroboros/np1_flow.h>
 #include <ouroboros/rib.h>
 #include <ouroboros/random.h>
 #include <ouroboros/pthread.h>
@@ -61,9 +62,10 @@
 #define TIMEOUT  10 * MILLION /* nanoseconds */
 #define MSGBUFSZ 32768
 
-#define FLOW_REQ    0
-#define FLOW_REPLY  1
-#define FLOW_UPDATE 2
+#define FLOW_REQ        0
+#define FLOW_REPLY      1
+#define FLOW_UPDATE     2
+#define FLOW_IRM_UPDATE 3
 
 #define STAT_FILE_LEN 0
 
@@ -585,6 +587,43 @@ static int fa_handle_flow_update(struct fa_msg * msg,
         return 0;
 }
 
+static int fa_handle_flow_irm_update(struct fa_msg * msg,
+                               size_t          len)
+{
+        buffer_t data;
+        int      fd;
+        int      flow_id;
+
+        if (len < sizeof(*msg))
+                return -EINVAL;
+
+        data.data = (uint8_t *) msg + sizeof(*msg);
+        data.len  = len - sizeof(*msg);
+
+        pthread_rwlock_rdlock(&fa.flows_lock);
+
+        fd = eid_to_fd(ntoh64(msg->r_eid));
+
+        pthread_rwlock_unlock(&fa.flows_lock);
+
+        if (fd < 0) {
+                log_err("Flow update for unknown EID %" PRIu64 ".",
+                        ntoh64(msg->r_eid));
+                return -ENOTALLOC;
+        }
+
+        flow_id = np1_flow_id(fd);
+        if (flow_id < 0)
+                return -ENOTALLOC;
+
+        if (ipcp_flow_update_arr(flow_id, &data) < 0) {
+                log_err("Failed to relay flow update on fd %d.", fd);
+                return -EIRMD;
+        }
+
+        return 0;
+}
+
 static void * fa_handle_packet(void * o)
 {
         (void) o;
@@ -613,6 +652,10 @@ static void * fa_handle_packet(void * o)
                         if (fa_handle_flow_update(msg, len) < 0)
                                 log_err("Error handling flow update.");
                         break;
+                case FLOW_IRM_UPDATE:
+                        if (fa_handle_flow_irm_update(msg, len) < 0)
+                                log_err("Error handling flow update.");
+                        break;
                 default:
                         log_warn("Recieved unknown flow allocation message.");
                         break;
@@ -872,6 +915,44 @@ int fa_alloc_resp(int              fd,
         return -1;
 }
 
+int fa_irm_update(int              fd,
+            const buffer_t * data)
+{
+        struct fa_msg *      msg;
+        struct ssm_pk_buff * spb;
+        struct fa_flow *     flow;
+        qoscube_t            qc = QOS_CUBE_BE;
+        uint64_t             r_addr;
+
+        flow = &fa.flows[fd];
+
+        if (ipcp_spb_reserve(&spb, sizeof(*msg) + data->len))
+                return -1;
+
+        msg = (struct fa_msg *) ssm_pk_buff_head(spb);
+        memset(msg, 0, sizeof(*msg));
+
+        msg->code     = FLOW_IRM_UPDATE;
+        if (data->len > 0)
+                memcpy(msg + 1, data->data, data->len);
+
+        pthread_rwlock_rdlock(&fa.flows_lock);
+
+        msg->r_eid = hton64(flow->r_eid);
+        msg->s_eid = hton64(flow->s_eid);
+        r_addr     = flow->r_addr;
+
+        pthread_rwlock_unlock(&fa.flows_lock);
+
+        if (dt_write_packet(r_addr, qc, fa.eid, spb)) {
+                log_err("Failed to send flow update packet.");
+                ipcp_spb_release(spb);
+                return -1;
+        }
+
+        return 0;
+}
+
 int fa_dealloc(int fd)
 {
         if (ipcp_flow_fini(fd) < 0)
diff --git a/src/ipcpd/unicast/fa.h b/src/ipcpd/unicast/fa.h
index 0c19dc25..f31b40e9 100644
--- a/src/ipcpd/unicast/fa.h
+++ b/src/ipcpd/unicast/fa.h
@@ -45,6 +45,9 @@ int  fa_alloc_resp(int              fd,
 
 int  fa_dealloc(int fd);
 
+int  fa_irm_update(int              fd,
+                   const buffer_t * data);
+
 void fa_np1_rcv(uint64_t             eid,
                 uint8_t              ecn,
                 struct ssm_pk_buff * spb);
diff --git a/src/ipcpd/unicast/main.c b/src/ipcpd/unicast/main.c
index 9a35531e..1155b88b 100644
--- a/src/ipcpd/unicast/main.c
+++ b/src/ipcpd/unicast/main.c
@@ -273,7 +273,8 @@ static struct ipcp_ops unicast_ops = {
         .ipcp_flow_alloc      = fa_alloc,
         .ipcp_flow_join       = NULL,
         .ipcp_flow_alloc_resp = fa_alloc_resp,
-        .ipcp_flow_dealloc    = fa_dealloc
+        .ipcp_flow_dealloc    = fa_dealloc,
+        .ipcp_flow_update     = fa_irm_update
 };
 
 int main(int    argc,
diff --git a/src/irmd/ipcp.c b/src/irmd/ipcp.c
index a7da186c..7eccfc80 100644
--- a/src/irmd/ipcp.c
+++ b/src/irmd/ipcp.c
@@ -444,6 +444,38 @@ int ipcp_flow_join(const struct flow_info * flow,
         return ret;
 }
 
+int ipcp_flow_update(const struct flow_info * flow,
+                     const buffer_t           data)
+{
+        ipcp_msg_t   msg = IPCP_MSG__INIT;
+        ipcp_msg_t * recv_msg;
+        int          ret;
+
+        msg.code         = IPCP_MSG_CODE__IPCP_FLOW_UPDATE;
+        msg.has_flow_id  = true;
+        msg.flow_id      = flow->id;
+        msg.has_pk       = true;
+        msg.pk.data      = data.data;
+        msg.pk.len       = data.len;
+
+        recv_msg = send_recv_ipcp_msg(flow->n_1_pid, &msg);
+        if (recv_msg == NULL) {
+                log_err("Did not receive message.");
+                return -EIPCP;
+        }
+
+        if (!recv_msg->has_result) {
+                log_err("Message has no result");
+                ipcp_msg__free_unpacked(recv_msg, NULL);
+                return -EIPCP;
+        }
+
+        ret = recv_msg->result;
+        ipcp_msg__free_unpacked(recv_msg, NULL);
+
+        return ret;
+}
+
 int ipcp_flow_alloc(const struct flow_info * flow,
                     const buffer_t           dst,
                     const buffer_t           data)
diff --git a/src/irmd/ipcp.h b/src/irmd/ipcp.h
index f1025096..8d06623c 100644
--- a/src/irmd/ipcp.h
+++ b/src/irmd/ipcp.h
@@ -68,4 +68,7 @@ int   ipcp_flow_dealloc(pid_t  pid,
                         int    flow_id,
                         time_t timeo);
 
+int   ipcp_flow_update(const struct flow_info * flow,
+                       const buffer_t           data);
+
 #endif /* OUROBOROS_IRMD_IPCP_H */
diff --git a/src/lib/dev.c b/src/lib/dev.c
index 543bd13e..cff1ebf2 100644
--- a/src/lib/dev.c
+++ b/src/lib/dev.c
@@ -2282,6 +2282,38 @@ int np1_flow_resp(int flow_id,
         return fd;
 }
 
+int np1_flow_fd(int flow_id)
+{
+        int fd;
+
+        if (flow_id < 0 || flow_id >= SYS_MAX_FLOWS)
+                return -1;
+
+        pthread_rwlock_rdlock(&proc.lock);
+
+        fd = proc.id_to_fd[flow_id].fd;
+
+        pthread_rwlock_unlock(&proc.lock);
+
+        return fd;
+}
+
+int np1_flow_id(int fd)
+{
+        int flow_id;
+
+        if (fd < 0 || fd >= PROC_MAX_FLOWS)
+                return -1;
+
+        pthread_rwlock_rdlock(&proc.lock);
+
+        flow_id = proc.flows[fd].info.id;
+
+        pthread_rwlock_unlock(&proc.lock);
+
+        return flow_id;
+}
+
 int ipcp_create_r(const struct ipcp_info * info)
 {
         uint8_t          buf[SOCK_BUF_SIZE];
@@ -2350,6 +2382,29 @@ int ipcp_flow_req_arr(const buffer_t * dst,
         return flow_init(&flow, &crypt, 0);
 }
 
+int ipcp_flow_update_arr(int              flow_id,
+                         const buffer_t * data)
+{
+        struct flow_info flow;
+        uint8_t          buf[SOCK_BUF_SIZE];
+        buffer_t         msg = {SOCK_BUF_SIZE, buf};
+        int              err;
+
+        memset(&flow, 0, sizeof(flow));
+
+        flow.id      = flow_id;
+        flow.n_1_pid = getpid();
+
+        if (ipcp_flow_update_arr__irm_req_ser(&msg, &flow, data) < 0)
+                return -ENOMEM;
+
+        err = send_recv_msg(&msg);
+        if (err < 0)
+                return err;
+
+        return irm__irm_result_des(&msg);
+}
+
 int ipcp_flow_alloc_reply(int              fd,
                           int              response,
                           time_t           mpl,
diff --git a/src/lib/pb/ipcp.proto b/src/lib/pb/ipcp.proto
index 406b8d9c..afee4f91 100644
--- a/src/lib/pb/ipcp.proto
+++ b/src/lib/pb/ipcp.proto
@@ -39,6 +39,7 @@ enum ipcp_msg_code {
         IPCP_CONNECT         = 10;
         IPCP_DISCONNECT      = 11;
         IPCP_REPLY           = 12;
+        IPCP_FLOW_UPDATE     = 13;
 }
 
 message ipcp_msg {
diff --git a/src/lib/pb/irm.proto b/src/lib/pb/irm.proto
index 5de860a5..98b75a95 100644
--- a/src/lib/pb/irm.proto
+++ b/src/lib/pb/irm.proto
@@ -53,6 +53,8 @@ enum irm_msg_code {
         IPCP_FLOW_REQ_ARR     = 25;
         IPCP_FLOW_ALLOC_REPLY = 26;
         IRM_REPLY             = 27;
+        IRM_FLOW_UPDATE       = 28;
+        IPCP_FLOW_UPDATE_ARR  = 29;
 }
 
 message timespec_msg {
@@ -96,4 +98,6 @@ message irm_msg {
         optional sint32 result           = 25;
         optional bytes  sym_key          = 26; /* symmetric encryption key */
         optional sint32 cipher_nid       = 27; /* cipher NID */
+        optional uint32 generation       = 28; /* re-key batch generation */
+        optional bool   rekey            = 29; /* re-key watermark trigger */
 }
diff --git a/src/lib/serdes-irm.c b/src/lib/serdes-irm.c
index 65f2c02d..a896576d 100644
--- a/src/lib/serdes-irm.c
+++ b/src/lib/serdes-irm.c
@@ -444,6 +444,56 @@ int ipcp_flow_alloc_reply__irm_msg_ser(buffer_t *               buf,
 
         return 0;
  fail_msg:
+        /* hash/pk are borrowed from the caller; detach before free. */
+        msg->hash.len  = 0;
+        msg->hash.data = NULL;
+        msg->pk.len    = 0;
+        msg->pk.data   = NULL;
+        irm_msg__free_unpacked(msg, NULL);
+ fail_malloc:
+        return -ENOMEM;
+}
+
+int ipcp_flow_update_arr__irm_req_ser(buffer_t *               buf,
+                                      const struct flow_info * flow,
+                                      const buffer_t *         data)
+{
+        irm_msg_t * msg;
+        size_t      len;
+
+        msg = malloc(sizeof(*msg));
+        if (msg == NULL)
+                goto fail_malloc;
+
+        irm_msg__init(msg);
+
+        msg->code         = IRM_MSG_CODE__IPCP_FLOW_UPDATE_ARR;
+        msg->flow_info    = flow_info_s_to_msg(flow);
+        if (msg->flow_info == NULL)
+                goto fail_msg;
+
+        msg->has_pk       = true;
+        msg->pk.len       = data->len;
+        msg->pk.data      = data->data;
+
+        len = irm_msg__get_packed_size(msg);
+        if (len == 0 || len > buf->len)
+                goto fail_msg;
+
+        buf->len = len;
+
+        irm_msg__pack(msg, buf->data);
+
+        /* Don't free data! */
+        msg->pk.len  = 0;
+        msg->pk.data = NULL;
+        irm_msg__free_unpacked(msg, NULL);
+
+        return 0;
+ fail_msg:
+        /* pk.data is borrowed from the caller; detach before free. */
+        msg->pk.len  = 0;
+        msg->pk.data = NULL;
         irm_msg__free_unpacked(msg, NULL);
  fail_malloc:
         return -ENOMEM;