From c386d9b7caa56f472fdce20ff5b2841ed41dd539 Mon Sep 17 00:00:00 2001 From: Dimitri Staessens Date: Sun, 14 Jun 2026 16:00:30 +0200 Subject: 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 Signed-off-by: Sander Vrijders --- src/ipcpd/eth/eth.c | 98 +++++++++++++++++++++++++++++++++++++++++++++++- src/ipcpd/ipcp.c | 34 +++++++++++++++++ src/ipcpd/ipcp.h | 3 ++ src/ipcpd/local/main.c | 36 +++++++++++++++++- src/ipcpd/udp/udp.c | 81 ++++++++++++++++++++++++++++++++++++++- src/ipcpd/unicast/fa.c | 87 ++++++++++++++++++++++++++++++++++++++++-- src/ipcpd/unicast/fa.h | 3 ++ src/ipcpd/unicast/main.c | 3 +- 8 files changed, 338 insertions(+), 7 deletions(-) (limited to 'src/ipcpd') 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 #include #include +#include #include #include #include @@ -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(ð_data.flows_lock); + + flow = ð_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(ð_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(ð_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(ð_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(ð_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 #include #include +#include #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 #include #include +#include +#include #include #include #include @@ -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 #include #include +#include #include #include #include @@ -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, -- cgit v1.2.3