diff options
Diffstat (limited to 'src/lib')
| -rw-r--r-- | src/lib/CMakeLists.txt | 17 | ||||
| -rw-r--r-- | src/lib/config.h.in | 19 | ||||
| -rw-r--r-- | src/lib/dev.c | 135 | ||||
| -rw-r--r-- | src/lib/irm.c | 127 | ||||
| -rw-r--r-- | src/lib/pb/ipcp.proto | 7 | ||||
| -rw-r--r-- | src/lib/pb/ipcp_config.proto | 27 | ||||
| -rw-r--r-- | src/lib/pb/irm.proto | 9 | ||||
| -rw-r--r-- | src/lib/pb/model.proto | 13 | ||||
| -rw-r--r-- | src/lib/poa/addr.c | 142 | ||||
| -rw-r--r-- | src/lib/poa/eth.c | 1987 | ||||
| -rw-r--r-- | src/lib/poa/poa.c | 2515 | ||||
| -rw-r--r-- | src/lib/poa/poa.h | 364 | ||||
| -rw-r--r-- | src/lib/poa/udp.c | 633 | ||||
| -rw-r--r-- | src/lib/protobuf.c | 386 | ||||
| -rw-r--r-- | src/lib/serdes-irm.c | 133 | ||||
| -rw-r--r-- | src/lib/tests/CMakeLists.txt | 7 | ||||
| -rw-r--r-- | src/lib/tests/poa_test.c | 307 |
17 files changed, 6686 insertions, 142 deletions
diff --git a/src/lib/CMakeLists.txt b/src/lib/CMakeLists.txt index 3abf39d0..48064ce4 100644 --- a/src/lib/CMakeLists.txt +++ b/src/lib/CMakeLists.txt @@ -95,9 +95,9 @@ endif() if(HAVE_LIBURCU) target_link_libraries(ouroboros-common PRIVATE Urcu::Urcu) - # urcu headers require C99; override the global -std=c89 for this TU only. - set_source_files_properties(crypt/keyrot.c PROPERTIES - COMPILE_OPTIONS "-std=gnu99") + # urcu headers require C99; override the global -std=c89 for these TUs. + set_source_files_properties(crypt/keyrot.c dev.c poa/addr.c poa/eth.c + poa/udp.c PROPERTIES COMPILE_OPTIONS "-std=gnu99") endif() install(TARGETS ouroboros-common @@ -107,8 +107,14 @@ install(TARGETS ouroboros-common set(SOURCE_FILES_DEV cep.c dev.c + poa/addr.c + poa/udp.c ) +if(HAVE_ETH) + list(APPEND SOURCE_FILES_DEV poa/eth.c) +endif() + add_library(ouroboros-dev SHARED ${SOURCE_FILES_DEV} ${CEP_PROTO_SRCS}) @@ -131,6 +137,11 @@ target_include_directories(ouroboros-dev target_link_libraries(ouroboros-dev PUBLIC ouroboros-common) +if(HAVE_LIBURCU) + # dev.c calls urcu directly; do not rely on transitive linkage. + target_link_libraries(ouroboros-dev PRIVATE Urcu::Urcu) +endif() + install(TARGETS ouroboros-dev EXPORT OuroborosTargets LIBRARY DESTINATION ${CMAKE_INSTALL_LIBDIR}) diff --git a/src/lib/config.h.in b/src/lib/config.h.in index 4ebaa194..3c6985c3 100644 --- a/src/lib/config.h.in +++ b/src/lib/config.h.in @@ -79,6 +79,25 @@ #define PROC_RES_FDS @PROC_RES_FDS@ #define PROC_MAX_FQUEUES @PROC_MAX_FQUEUES@ +/* Flow endpoints */ +#define POA_MGMT_FRAME_SIZE @POA_MGMT_FRAME_SIZE@ +#define POA_MGMT_SND_TIMEO @POA_MGMT_SND_TIMEO@ +#define POA_MAX_POAS @POA_MAX_POAS@ +#define POA_UDP_MPL @POA_UDP_MPL@ +#define POA_UDP4_MTU @POA_UDP4_MTU@ +#define POA_UDP6_MTU @POA_UDP6_MTU@ +#define POA_UDP_RD_BUF @POA_UDP_RD_BUF@ +#cmakedefine HAVE_ETH +#cmakedefine HAVE_RAW_SOCKETS +#cmakedefine HAVE_BPF +#cmakedefine HAVE_NETMAP +#cmakedefine POA_ETH_QDISC_BYPASS +#define POA_ETH_MPL @POA_ETH_MPL@ +#define POA_ETH_SNDBUF @POA_ETH_SNDBUF@ +#define POA_ETH_RCVBUF @POA_ETH_RCVBUF@ +#define POA_ETH_LO_MTU @POA_ETH_LO_MTU@ +#define POA_ETH_RD_BUF @POA_ETH_RD_BUF@ + /* Default Delta-t parameters */ #cmakedefine FRCT_LINUX_RTT_ESTIMATOR #define DELT_A (@DELTA_T_ACK@) /* ms */ diff --git a/src/lib/dev.c b/src/lib/dev.c index 3fb8d831..eb706691 100644 --- a/src/lib/dev.c +++ b/src/lib/dev.c @@ -28,6 +28,9 @@ #include "config.h" #include "ssm.h" +#include "poa/poa.h" + +#define OUROBOROS_PREFIX "libouroboros" #include <ouroboros/atomics.h> #include <ouroboros/bitmap.h> @@ -45,6 +48,7 @@ #include <ouroboros/ipcp-dev.h> #include <ouroboros/list.h> #include <ouroboros/local-dev.h> +#include <ouroboros/logs.h> #include <ouroboros/np1_flow.h> #include <ouroboros/pthread.h> #include <ouroboros/random.h> @@ -63,8 +67,10 @@ #ifdef HAVE_LIBGCRYPT #include <gcrypt.h> #endif +#include <arpa/inet.h> #include <stdarg.h> #include <stdbool.h> +#include <stddef.h> #include <inttypes.h> #include <stdio.h> #include <stdlib.h> @@ -98,8 +104,8 @@ struct flow { ssize_t part_idx; struct crypt_ctx * crypt; - int headsz; /* Selector */ - int tailsz; /* Tag + CRC */ + int headsz; /* Selector */ + int tailsz; /* Tag + CRC */ struct timespec rk_grace; /* TX-promote deadline */ struct timespec rk_attempt; /* Last re-key attempt */ @@ -116,6 +122,8 @@ struct flow { struct timespec rcv_timeo; struct frcti * frcti; + + struct poa_flow * poa; /* NULL for shared memory flows */ }; struct flow_set { @@ -696,6 +704,9 @@ static void do_flow_fini(int fd) { assert(fd >= 0 && fd < PROC_MAX_FLOWS); + if (proc.flows[fd].poa != NULL) + poa_flow_detach(proc.flows[fd].poa); + if (proc.flows[fd].frcti != NULL) frcti_destroy(proc.flows[fd].frcti); @@ -750,9 +761,11 @@ static __inline__ size_t flow_user_mtu(const struct flow * flow, return raw > hdr ? raw - hdr : 0; } +/* A PoA flow transmits on its own socket; it has no tx ring. */ static int flow_init(struct flow_info * info, struct crypt_sk * sk, - time_t rtt_hint) + time_t rtt_hint, + struct poa_flow * pf) { struct timespec now; struct flow * flow; @@ -777,13 +790,15 @@ static int flow_init(struct flow_info * info, if (flow->rx_rb == NULL) goto fail_rx_rb; - flow->tx_rb = ssm_rbuff_open(info->n_1_pid, info->id); - if (flow->tx_rb == NULL) - goto fail_tx_rb; + if (pf == NULL) { + flow->tx_rb = ssm_rbuff_open(info->n_1_pid, info->id); + if (flow->tx_rb == NULL) + goto fail_tx_rb; - flow->set = ssm_flow_set_open(info->n_1_pid); - if (flow->set == NULL) - goto fail_set; + flow->set = ssm_flow_set_open(info->n_1_pid); + if (flow->set == NULL) + goto fail_set; + } flow->oflags = FLOWFDEFAULT; flow->part_idx = NO_PART; @@ -792,11 +807,14 @@ static int flow_init(struct flow_info * info, flow->crypt = NULL; flow->headsz = 0; flow->tailsz = 0; + flow->poa = pf; if (IS_ENCRYPTED(sk)) { flow->crypt = crypt_create_ctx(sk); - if (flow->crypt == NULL) + if (flow->crypt == NULL) { + err = -ECRYPT; goto fail_crypt; + } flow->headsz = crypt_get_headsz(flow->crypt); flow->tailsz = crypt_get_tagsz(flow->crypt); } @@ -815,6 +833,9 @@ static int flow_init(struct flow_info * info, proc.id_to_fd[info->id].fd = fd; + if (pf != NULL) + poa_flow_attach(pf, info->id, flow->rx_rb); + flow_set_state(&proc.id_to_fd[info->id], FLOW_ALLOCATED); pthread_rwlock_unlock(&proc.lock); @@ -824,9 +845,11 @@ static int flow_init(struct flow_info * info, fail_frcti: crypt_destroy_ctx(flow->crypt); fail_crypt: - ssm_flow_set_close(flow->set); + if (flow->set != NULL) + ssm_flow_set_close(flow->set); fail_set: - ssm_rbuff_close(flow->tx_rb); + if (flow->tx_rb != NULL) + ssm_rbuff_close(flow->tx_rb); fail_tx_rb: ssm_rbuff_close(flow->rx_rb); fail_rx_rb: @@ -1077,14 +1100,20 @@ static void fini(void) __attribute__((section(INIT_SECTION))) __typeof__(init) * __init = init; __attribute__((section(FINI_SECTION))) __typeof__(fini) * __fini = fini; +/* + * A PoA flow is announced by its peer before the accept; from the + * reply on, the peer may transmit, so the flow must be able to + * receive. + */ int flow_accept(qosspec_t * qs, const struct timespec * timeo) { struct flow_info flow; - struct crypt_sk crypt; + struct crypt_sk crypt; uint8_t buf[SOCK_BUF_SIZE]; buffer_t msg = {SOCK_BUF_SIZE, buf}; uint8_t key[SYMMKEYSZ]; + struct poa_flow * pf; int fd; int err; @@ -1116,8 +1145,13 @@ int flow_accept(qosspec_t * qs, if (err < 0) return err; - /* No RTT in accept; rtt_hint=0 bootstraps from first ACK. */ - fd = flow_init(&flow, &crypt, 0); + pf = poa_flow_take_pending(flow.id); + + fd = flow_init(&flow, &crypt, 0, pf); + if (fd >= 0) + poa_flow_ready(pf); + else if (pf != NULL) + poa_flow_detach(pf); crypt_secure_clear(key, SYMMKEYSZ); @@ -1173,7 +1207,7 @@ int flow_alloc(const char * dst, if (err < 0) return err; - fd = flow_init(&flow, &crypt, ts_diff_ns(&t1, &t0)); + fd = flow_init(&flow, &crypt, ts_diff_ns(&t1, &t0), NULL); crypt_secure_clear(key, SYMMKEYSZ); @@ -1214,7 +1248,7 @@ int flow_join(const char * dst, if (err < 0) return err; - fd = flow_init(&flow, &crypt, 0); + fd = flow_init(&flow, &crypt, 0, NULL); crypt_secure_clear(key, SYMMKEYSZ); @@ -1275,7 +1309,8 @@ int flow_dealloc(int fd) pthread_cleanup_push(__cleanup_rwlock_unlock, &proc.lock); - ssm_rbuff_fini(flow->tx_rb); + if (flow->tx_rb != NULL) + ssm_rbuff_fini(flow->tx_rb); pthread_cleanup_pop(true); @@ -1428,7 +1463,11 @@ int fccntl(int fd, break; case FLOWGTXQLEN: qlen = va_arg(l, size_t *); - *qlen = ssm_rbuff_queued(flow->tx_rb); + + if (flow->poa != NULL) + *qlen = poa_flow_qpkts(flow->poa); + else + *qlen = ssm_rbuff_queued(flow->tx_rb); break; case FLOWGMTU: maxp = va_arg(l, size_t *); @@ -1456,16 +1495,18 @@ int fccntl(int fd, if (flow->oflags & FLOWFDOWN) { ssm_rbuff_set_bits(flow->rx_rb, RB_FLOWDOWN); - ssm_rbuff_set_bits(flow->tx_rb, RB_FLOWDOWN); - ssm_flow_set_notify(flow->set, - flow->info.id, - FLOW_DOWN); + if (flow->tx_rb != NULL) + ssm_rbuff_set_bits(flow->tx_rb, RB_FLOWDOWN); + if (flow->set != NULL) + ssm_flow_set_notify(flow->set, flow->info.id, + FLOW_DOWN); } else { ssm_rbuff_clr_bits(flow->rx_rb, RB_FLOWDOWN); - ssm_rbuff_clr_bits(flow->tx_rb, RB_FLOWDOWN); - ssm_flow_set_notify(flow->set, - flow->info.id, - FLOW_UP); + if (flow->tx_rb != NULL) + ssm_rbuff_clr_bits(flow->tx_rb, RB_FLOWDOWN); + if (flow->set != NULL) + ssm_flow_set_notify(flow->set, flow->info.id, + FLOW_UP); } break; @@ -1599,6 +1640,9 @@ static int flow_tx_spb(struct flow * flow, goto enomem; } + if (flow->poa != NULL) + return poa_flow_tx(flow->poa, spb, block, abstime); + if (!block) ret = ssm_rbuff_write(flow->tx_rb, idx); else @@ -2450,7 +2494,7 @@ int np1_flow_alloc(pid_t n_pid, /* np1 flow: n_1_pid is the upper. */ flow.n_1_pid = n_pid; - return flow_init(&flow, &crypt, 0); + return flow_init(&flow, &crypt, 0, NULL); } int np1_flow_dealloc(int flow_id, @@ -2584,7 +2628,7 @@ int ipcp_flow_req_arr(const buffer_t * dst, crypt.nid = NID_undef; - return flow_init(&flow, &crypt, 0); + return flow_init(&flow, &crypt, 0, NULL); } int ipcp_flow_update_arr(int flow_id, @@ -2853,11 +2897,13 @@ int ipcp_flow_fini(int fd) } ssm_rbuff_set_bits(proc.flows[fd].rx_rb, RB_FLOWDOWN); - ssm_rbuff_set_bits(proc.flows[fd].tx_rb, RB_FLOWDOWN); - ssm_flow_set_notify(proc.flows[fd].set, - proc.flows[fd].info.id, - FLOW_DEALLOC); + if (proc.flows[fd].tx_rb != NULL) + ssm_rbuff_set_bits(proc.flows[fd].tx_rb, RB_FLOWDOWN); + + if (proc.flows[fd].set != NULL) + ssm_flow_set_notify(proc.flows[fd].set, proc.flows[fd].info.id, + FLOW_DEALLOC); rx_rb = proc.flows[fd].rx_rb; @@ -2892,9 +2938,30 @@ size_t ipcp_flow_queued(int fd) assert(fd >= 0 && fd < PROC_MAX_FLOWS); assert(proc.flows[fd].info.id >= 0); + if (proc.flows[fd].poa != NULL) + return poa_flow_qlen(proc.flows[fd].poa); + return ssm_rbuff_queued(proc.flows[fd].tx_rb); } +int ipcp_flow_queue_id(int fd) +{ + int qid; + + assert(fd >= 0 && fd < PROC_MAX_FLOWS); + assert(proc.flows[fd].info.id >= 0); + + if (proc.flows[fd].poa == NULL) + return fd; + + /* An unidentified PoA answers for itself, never for an fd. */ + qid = poa_flow_qid(proc.flows[fd].poa); + if (qid < 0 || qid >= POA_MAX_POAS) + return fd; + + return PROC_MAX_FLOWS + qid; +} + int local_flow_transfer(int src_fd, int dst_fd, struct ssm_pool * src_pool, @@ -2960,3 +3027,5 @@ int local_flow_transfer(int src_fd, return ret; } + +#include "poa/poa.c" diff --git a/src/lib/irm.c b/src/lib/irm.c index c62701aa..5d53841f 100644 --- a/src/lib/irm.c +++ b/src/lib/irm.c @@ -118,6 +118,8 @@ int irm_bootstrap_ipcp(pid_t pid, msg.has_pid = true; msg.pid = pid; msg.conf = ipcp_config_s_to_msg(conf); + if (msg.conf == NULL) + return -ENOMEM; recv_msg = send_recv_irm_msg(&msg); ipcp_config_msg__free_unpacked(msg.conf, NULL); @@ -135,10 +137,11 @@ int irm_bootstrap_ipcp(pid_t pid, return ret; } -int irm_connect_ipcp(pid_t pid, - const char * dst, - const char * component, - qosspec_t qs) +int irm_connect_ipcp(pid_t pid, + const char * dst, + const char * component, + qosspec_t qs, + const struct poa_addr * addr) { irm_msg_t msg = IRM_MSG__INIT; irm_msg_t * recv_msg; @@ -152,7 +155,18 @@ int irm_connect_ipcp(pid_t pid, msg.pid = pid; msg.qosspec = qos_spec_s_to_msg(&qs); + if (addr != NULL) { + msg.peer = poa_addr_s_to_msg(addr); + if (msg.peer == NULL) { + qosspec_msg__free_unpacked(msg.qosspec, NULL); + return -ENOMEM; + } + } + recv_msg = send_recv_irm_msg(&msg); + + if (msg.peer != NULL) + poa_addr_msg__free_unpacked(msg.peer, NULL); qosspec_msg__free_unpacked(msg.qosspec, NULL); if (recv_msg == NULL) @@ -245,8 +259,99 @@ ssize_t irm_list_ipcps(struct ipcp_list_info ** ipcps) return nr; } -int irm_enroll_ipcp(pid_t pid, - const char * dst) +static int irm_attach_detach(pid_t pid, + const struct poa_spec * poa, + int code) +{ + irm_msg_t msg = IRM_MSG__INIT; + irm_msg_t * recv_msg; + int ret; + + if (pid == -1 || poa == NULL) + return -EINVAL; + + msg.code = code; + msg.has_pid = true; + msg.pid = pid; + msg.poa = poa_spec_s_to_msg(poa); + if (msg.poa == NULL) + return -EINVAL; + + recv_msg = send_recv_irm_msg(&msg); + + poa_spec_msg__free_unpacked(msg.poa, NULL); + + if (recv_msg == NULL) + return -EIRMD; + + if (!recv_msg->has_result) { + irm_msg__free_unpacked(recv_msg, NULL); + return -EIRMD; + } + + ret = recv_msg->result; + + irm_msg__free_unpacked(recv_msg, NULL); + + return ret; +} + +ssize_t irm_list_poas(pid_t pid, + struct poa_spec ** poas) +{ + irm_msg_t msg = IRM_MSG__INIT; + irm_msg_t * recv_msg; + size_t nr; + size_t i; + + if (pid == -1 || poas == NULL) + return -EINVAL; + + *poas = NULL; + + msg.code = IRM_MSG_CODE__IRM_LIST_POAS; + msg.has_pid = true; + msg.pid = pid; + + recv_msg = send_recv_irm_msg(&msg); + if (recv_msg == NULL) + return -EIRMD; + + nr = recv_msg->n_poas; + if (nr == 0) { + irm_msg__free_unpacked(recv_msg, NULL); + return 0; + } + + *poas = malloc(nr * sizeof(**poas)); + if (*poas == NULL) { + irm_msg__free_unpacked(recv_msg, NULL); + return -ENOMEM; + } + + for (i = 0; i < nr; i++) + (*poas)[i] = poa_spec_msg_to_s(recv_msg->poas[i]); + + irm_msg__free_unpacked(recv_msg, NULL); + + return (ssize_t) nr; +} + +int irm_attach_ipcp(pid_t pid, + const struct poa_spec * poa) +{ + return irm_attach_detach(pid, poa, IRM_MSG_CODE__IRM_ATTACH_IPCP); +} + +int irm_detach_ipcp(pid_t pid, + const struct poa_spec * poa) +{ + return irm_attach_detach(pid, poa, IRM_MSG_CODE__IRM_DETACH_IPCP); +} + +int irm_enroll_ipcp(pid_t pid, + const char * dst, + const struct poa_addr * addr) { irm_msg_t msg = IRM_MSG__INIT; irm_msg_t * recv_msg = NULL; @@ -260,7 +365,17 @@ int irm_enroll_ipcp(pid_t pid, msg.pid = pid; msg.dst = (char *) dst; + if (addr != NULL) { + msg.peer = poa_addr_s_to_msg(addr); + if (msg.peer == NULL) + return -ENOMEM; + } + recv_msg = send_recv_irm_msg(&msg); + + if (msg.peer != NULL) + poa_addr_msg__free_unpacked(msg.peer, NULL); + if (recv_msg == NULL) return -EIRMD; diff --git a/src/lib/pb/ipcp.proto b/src/lib/pb/ipcp.proto index afee4f91..298a15e2 100644 --- a/src/lib/pb/ipcp.proto +++ b/src/lib/pb/ipcp.proto @@ -40,6 +40,9 @@ enum ipcp_msg_code { IPCP_DISCONNECT = 11; IPCP_REPLY = 12; IPCP_FLOW_UPDATE = 13; + IPCP_ATTACH = 14; + IPCP_DETACH = 15; + IPCP_LIST_POAS = 16; } message ipcp_msg { @@ -58,4 +61,8 @@ message ipcp_msg { optional sint32 mpl = 13; /* MPL in ms. */ optional int32 result = 14; optional uint32 uid = 15; /* 0 = GSPP, >0 = PUP uid */ + optional poa_addr_msg peer = 16; /* peer PoA address */ + optional bool is_poa = 17; /* flow rides a PoA */ + optional poa_spec_msg poa = 18; /* PoA to attach */ + repeated poa_spec_msg poas = 19; /* PoAs attached */ } diff --git a/src/lib/pb/ipcp_config.proto b/src/lib/pb/ipcp_config.proto index a111b6d8..f6d313ac 100644 --- a/src/lib/pb/ipcp_config.proto +++ b/src/lib/pb/ipcp_config.proto @@ -66,29 +66,32 @@ message uni_config_msg { required uint32 cong_avoid = 4; } -message eth_config_msg { +message eth_poa_msg { required string dev = 1; required uint32 ethertype = 2; + optional bytes mac = 3; } -message udp4_config_msg { - required uint32 ip_addr = 1; - required uint32 port = 2; - required uint32 dns_addr = 3; /* set to 0 if unused */ +message udp4_poa_msg { + required uint32 ip_addr = 1; + required uint32 port = 2; } -message udp6_config_msg { - required bytes ip_addr = 1; - required uint32 port = 2; - required bytes dns_addr = 3; /* set to NULL if unused */ +message udp6_poa_msg { + required bytes ip_addr = 1; + required uint32 port = 2; } +/* Exactly one field is set; presence is the PoA type. */ +message poa_spec_msg { + optional udp4_poa_msg udp4 = 1; + optional udp6_poa_msg udp6 = 2; + optional eth_poa_msg eth = 3; +} + message ipcp_config_msg { required layer_info_msg layer_info = 1; required uint32 ipcp_type = 2; optional uni_config_msg unicast = 3; - optional udp4_config_msg udp4 = 4; - optional udp6_config_msg udp6 = 5; - optional eth_config_msg eth = 6; } diff --git a/src/lib/pb/irm.proto b/src/lib/pb/irm.proto index f54bc9ea..8f594fa3 100644 --- a/src/lib/pb/irm.proto +++ b/src/lib/pb/irm.proto @@ -55,6 +55,12 @@ enum irm_msg_code { IRM_REPLY = 27; IRM_FLOW_UPDATE = 28; IPCP_FLOW_UPDATE_ARR = 29; + IRM_POA_FLOW_ALLOC = 30; + IRM_POA_FLOW_ALLOC_R = 31; + IPCP_POA_FLOW_REQ_ARR = 32; + IRM_ATTACH_IPCP = 33; + IRM_DETACH_IPCP = 34; + IRM_LIST_POAS = 35; } message timespec_msg { @@ -101,4 +107,7 @@ message irm_msg { optional uint32 generation = 28; /* re-key batch generation */ optional bool rekey = 29; /* re-key watermark trigger */ optional bool rk_initiator = 30; /* re-key proof-holder side */ + optional poa_addr_msg peer = 32; /* peer PoA address */ + optional poa_spec_msg poa = 33; /* PoA to attach */ + repeated poa_spec_msg poas = 34; /* PoAs attached */ } diff --git a/src/lib/pb/model.proto b/src/lib/pb/model.proto index 4c1564a5..e00f3e12 100644 --- a/src/lib/pb/model.proto +++ b/src/lib/pb/model.proto @@ -58,6 +58,19 @@ message layer_info_msg { required uint32 dir_hash_algo = 2; } +/* Address of a flow PoA peer. */ +message poa_addr_msg { + required uint32 type = 1; /* enum poa_type */ + optional uint32 ip4 = 2; + optional bytes ip6 = 3; + optional uint32 port = 4; + optional bytes mac = 5; + optional uint32 ethertype = 6; + optional string dev = 7; /* our device, eth only */ + optional string hostname = 8; /* IRMd resolves, UDP only */ + optional bytes src_mac = 9; /* our MAC, eth only */ +} + message ipcp_info_msg { required uint32 type = 1; required string name = 2; diff --git a/src/lib/poa/addr.c b/src/lib/poa/addr.c new file mode 100644 index 00000000..d8811244 --- /dev/null +++ b/src/lib/poa/addr.c @@ -0,0 +1,142 @@ +/* + * Ouroboros - Copyright (C) 2016 - 2026 + * + * Points of attachment (PoA) - addresses and management messages + * + * Dimitri Staessens <dimitri@ouroboros.rocks> + * Sander Vrijders <sander@ouroboros.rocks> + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public License + * version 2.1 as published by the Free Software Foundation. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., http://www.fsf.org/about/contact/. + */ + +#if defined(__linux__) || defined(__CYGWIN__) +#ifndef _DEFAULT_SOURCE /* Test include source */ +#define _DEFAULT_SOURCE +#endif +#endif + +#include "config.h" + +#include <ouroboros/endian.h> +#include <ouroboros/errno.h> + +#include "poa.h" + +#include <arpa/inet.h> +#include <stdio.h> +#include <string.h> + +int poa_addr_cmp(const struct poa_addr * a, + const struct poa_addr * b) +{ + if (a->type != b->type) + return 1; + + switch (a->type) { + case POA_UDP4: + if (a->udp4.port != b->udp4.port) + return 1; + return memcmp(&a->udp4.ip_addr, &b->udp4.ip_addr, + sizeof(a->udp4.ip_addr)); + case POA_UDP6: + if (a->udp6.port != b->udp6.port) + return 1; + return memcmp(&a->udp6.ip_addr, &b->udp6.ip_addr, + sizeof(a->udp6.ip_addr)); + case POA_ETH: + if (a->eth.dst.ethertype != b->eth.dst.ethertype) + return 1; + return memcmp(a->eth.dst.mac, b->eth.dst.mac, POA_MAC_SIZE); + default: + return 1; + } +} + +/* Display/RIB entry name, e.g. "udp4.<ip>.<port>". */ +int poa_addr_name(const struct poa_addr * a, + char * buf, + size_t len) +{ + char ip[INET6_ADDRSTRLEN]; + int ret; + + switch (a->type) { + case POA_UDP4: + if (inet_ntop(AF_INET, &a->udp4.ip_addr, + ip, sizeof(ip)) == NULL) + return -EINVAL; + ret = snprintf(buf, len, "udp4.%s.%u", ip, a->udp4.port); + break; + case POA_UDP6: + if (inet_ntop(AF_INET6, &a->udp6.ip_addr, + ip, sizeof(ip)) == NULL) + return -EINVAL; + ret = snprintf(buf, len, "udp6.%s.%u", ip, a->udp6.port); + break; + case POA_ETH: + ret = snprintf(buf, len, "eth.%s.%04X", + a->eth.src.dev, a->eth.src.ethertype); + break; + default: + return -EINVAL; + } + + if (ret < 0 || (size_t) ret >= len) + return -EMSGSIZE; + + return 0; +} + +void poa_mgmt_msg_ser(struct poa_mgmt_msg * msg, + uint8_t code, + uint32_t s_eid, + uint32_t d_eid, + qosspec_t qs, + int response, + size_t data_len) +{ + memset(msg, 0, sizeof(*msg)); + + msg->code = code; + msg->s_eid = hton32(s_eid); + msg->d_eid = hton32(d_eid); + msg->response = hton32(response); + msg->data_len = hton16((uint16_t) data_len); + + if (code != POA_FLOW_REQ) + return; + + msg->bandwidth = hton64(qs.bandwidth); + msg->delay = hton32(qs.delay); + msg->loss = hton32(qs.loss); + msg->ber = hton32(qs.ber); + msg->max_gap = hton32(qs.max_gap); + msg->timeout = hton32(qs.timeout); + msg->availability = qs.availability; + msg->service = qs.service; +} + +void poa_mgmt_msg_qos(const struct poa_mgmt_msg * msg, + qosspec_t * qs) +{ + qs->bandwidth = ntoh64(msg->bandwidth); + qs->delay = ntoh32(msg->delay); + qs->loss = ntoh32(msg->loss); + qs->ber = ntoh32(msg->ber); + qs->max_gap = ntoh32(msg->max_gap); + qs->timeout = ntoh32(msg->timeout); + qs->availability = msg->availability; + qs->service = msg->service; +} + diff --git a/src/lib/poa/eth.c b/src/lib/poa/eth.c new file mode 100644 index 00000000..5e959ef6 --- /dev/null +++ b/src/lib/poa/eth.c @@ -0,0 +1,1987 @@ +/* + * Ouroboros - Copyright (C) 2016 - 2026 + * + * Points of attachment (PoA) - Ethernet transport + * + * Dimitri Staessens <dimitri@ouroboros.rocks> + * Sander Vrijders <sander@ouroboros.rocks> + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public License + * version 2.1 as published by the Free Software Foundation. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., http://www.fsf.org/about/contact/. + */ + +#if defined(__APPLE__) +#define _BSD_SOURCE +#define _DARWIN_C_SOURCE +#elif defined(__FreeBSD__) +#define __BSD_VISIBLE 1 +#elif defined(__linux__) || defined(__CYGWIN__) +#ifndef _DEFAULT_SOURCE +#define _DEFAULT_SOURCE +#endif +#else +#ifndef _POSIX_C_SOURCE +#define _POSIX_C_SOURCE 200809L +#endif +#endif + +#include "config.h" + +#define OUROBOROS_PREFIX "poa-eth" + +#include <ouroboros/endian.h> +#include <ouroboros/errno.h> +#include <ouroboros/hash.h> +#include <ouroboros/logs.h> +#include <ouroboros/pthread.h> +#include <ouroboros/sockets.h> +#include <ouroboros/time.h> + +#include "poa.h" + +#ifdef HAVE_RAW_SOCKETS +#include <net/if.h> +#include <netinet/in.h> +#include <linux/if_ether.h> +#include <linux/if_packet.h> +#include <linux/netlink.h> +#include <linux/gen_stats.h> +#include <linux/pkt_sched.h> +#include <linux/rtnetlink.h> +#include <linux/sockios.h> +#include <sys/ioctl.h> +#include <sys/socket.h> +#include <sys/uio.h> + +#include <ifaddrs.h> +#elif defined(HAVE_BPF) +#include <net/bpf.h> +#include <net/if.h> +#include <net/if_dl.h> +#include <sys/ioctl.h> +#include <sys/socket.h> +#include <sys/uio.h> + +#include <fcntl.h> +#include <ifaddrs.h> +#elif defined(HAVE_NETMAP) +#define NETMAP_WITH_LIBS +#include <net/netmap_user.h> +#include <net/if.h> +#include <sys/ioctl.h> +#include <sys/socket.h> + +#include <poll.h> +#ifndef __linux__ +#include <net/if_dl.h> +#include <ifaddrs.h> +#endif +#endif + +#include <errno.h> +#include <stdio.h> +#include <stdlib.h> +#include <string.h> +#include <unistd.h> + +#define ETH_TYPE_LEN_SIZE sizeof(uint16_t) +#define ETH_HDR_SIZE (2 * POA_MAC_SIZE + ETH_TYPE_LEN_SIZE) +#define POA_HDR_SIZE (sizeof(struct poa_hdr)) +#define ETH_HDR_TOT_SIZE (ETH_HDR_SIZE + POA_HDR_SIZE) + +#define BPF_DEV_MAX 256 /* /dev/bpfN units to try */ +/* Wait for the link to come back before reading it again. */ +#define ETH_DOWN_TIMEO 100 /* ms */ +/* Budget for a qdisc dump; the send path must not wait on netlink. */ +#define POA_QDISC_TIMEO 5 /* ms */ + +#ifndef ETH_MAX_MTU /* In if_ether.h as of Linux 4.10. */ +#define ETH_MAX_MTU 0xFFFFU +#endif + +struct poa_hdr { + uint16_t eid; + uint16_t len; + uint8_t hcs; +} __attribute__((packed)); + +struct eth_hdr { + uint8_t dst[POA_MAC_SIZE]; + uint8_t src[POA_MAC_SIZE]; + uint16_t ethertype; + struct poa_hdr poa; +} __attribute__((packed)); + + +struct eth_priv { + struct poa * poa; + + int s_fd; /* raw socket or bpf device */ + uint16_t ethertype; /* network order */ + uint8_t hw_addr[POA_MAC_SIZE]; + uint32_t mtu; /* device payload capacity */ + int ifindex; /* link id; 0 where unknown */ + + pthread_t reader; + bool running; +#ifdef HAVE_RAW_SOCKETS + struct sockaddr_ll device; +#elif defined(HAVE_BPF) + size_t blen; /* bpf device buffer size */ +#elif defined(HAVE_NETMAP) + struct nm_desc * nmd; + struct pollfd poll_in; + struct pollfd poll_out; +#endif + /* The kernel zeroes its counters on read, so accumulate. */ + size_t kern_rcv; + size_t kern_drp; +#ifdef HAVE_RAW_SOCKETS + /* Qdisc depth: netlink descriptor, its try-lock and cache. */ + int nl_fd; + uint8_t nl_busy; + size_t nl_pkt; + size_t nl_byt; + uint64_t nl_time; +#endif +}; + +struct eth_query { + struct list_head next; + uint8_t hash[POA_QUERY_HLEN]; + /* The resolve's constraints; replies must satisfy them. */ + uint16_t ethertype; + char c_dev[DEV_NAME_SIZE + 1]; + uint8_t c_mac[POA_MAC_SIZE]; + bool replied; + char dev[DEV_NAME_SIZE + 1]; + uint8_t src_mac[POA_MAC_SIZE]; + uint8_t mac[POA_MAC_SIZE]; + uint16_t r_ethertype; +}; + +static struct { + pthread_once_t once; + + struct llist pending; + pthread_mutex_t mtx; + pthread_cond_t cond; +} queries = { .once = PTHREAD_ONCE_INIT }; + +static void eth_hdr_ser(const struct eth_priv * priv, + struct eth_hdr * hdr, + const uint8_t * dst, + uint32_t eid, + size_t len) +{ + memcpy(hdr->dst, dst, POA_MAC_SIZE); + memcpy(hdr->src, priv->hw_addr, POA_MAC_SIZE); + + hdr->ethertype = priv->ethertype; + hdr->poa.eid = htons((uint16_t) eid); + hdr->poa.len = htons((uint16_t) len); + mem_hash(HASH_CRC8, &hdr->poa.hcs, + (uint8_t *) &hdr->poa.eid, 2 * sizeof(uint16_t)); +} + +/* Oriented from us to the sender: our PoA, then their MAC. */ +static void frame_to_addr(const struct eth_priv * priv, + const struct eth_hdr * hdr, + struct poa_addr * addr) +{ + memset(addr, 0, sizeof(*addr)); + + addr->type = priv->poa->type; + + addr->eth.src = priv->poa->local.eth.src; + + memcpy(addr->eth.dst.mac, hdr->src, POA_MAC_SIZE); + + addr->eth.dst.ethertype = ntohs(priv->ethertype); +} + +static bool frame_is_for_us(const struct eth_priv * priv, + const uint8_t * dst) +{ + static const uint8_t bc[POA_MAC_SIZE] = + { 0xff, 0xff, 0xff, 0xff, 0xff, 0xff }; + + if (memcmp(dst, priv->hw_addr, POA_MAC_SIZE) == 0) + return true; + + return memcmp(dst, bc, POA_MAC_SIZE) == 0; +} + +/* + * Parse and validate a frame; on success *eid and *plen locate the + * payload. Errors are silent: anyone can spray an interface. + */ +static int frame_parse(const struct eth_priv * priv, + const uint8_t * buf, + size_t len, + uint32_t * eid, + size_t * plen) +{ + const struct eth_hdr * hdr = (const struct eth_hdr *) buf; + uint8_t hcs; + + if (len < ETH_HDR_TOT_SIZE) + return -1; + + if (!frame_is_for_us(priv, hdr->dst)) + return -1; + + if (hdr->ethertype != priv->ethertype) + return -1; + + mem_hash(HASH_CRC8, &hcs, + (const uint8_t *) &hdr->poa.eid, 2 * sizeof(uint16_t)); + + if (hcs != hdr->poa.hcs) + return -1; + + *eid = ntohs(hdr->poa.eid); + + *plen = ntohs(hdr->poa.len); + if (*plen > len - ETH_HDR_TOT_SIZE) + return -1; + + return 0; +} + +#define ETH_QUERY_TIMEO 1900 /* ms total budget */ +#define ETH_QUERY_RETRIES 3 /* retransmits, 4 attempts total */ + +/* A resolve in progress; a reply fills every matching entry. */ +static void queries_init(void) +{ + pthread_condattr_t cattr; + + llist_init(&queries.pending); + + pthread_mutex_init(&queries.mtx, NULL); + + pthread_condattr_init(&cattr); +#ifndef __APPLE__ + pthread_condattr_setclock(&cattr, PTHREAD_COND_CLOCK); +#endif + pthread_cond_init(&queries.cond, &cattr); + pthread_condattr_destroy(&cattr); +} + +static void eth_query_cleanup(void * o) +{ + struct eth_query * q = (struct eth_query *) o; + + pthread_mutex_lock(&queries.mtx); + llist_del(&q->next, &queries.pending); + pthread_mutex_unlock(&queries.mtx); +} + +static void eth_query_arr(struct poa * poa, + const struct poa_addr * src, + const uint8_t * hash) +{ + uint8_t buf[sizeof(struct poa_mgmt_msg) + + POA_QUERY_HLEN]; + struct poa_mgmt_msg * msg = (struct poa_mgmt_msg *) buf; + + if (!poa_has_name(hash)) + return; + + poa_mgmt_msg_ser(msg, POA_NAME_REPLY, 0, 0, qos_raw, 0, POA_QUERY_HLEN); + + memcpy(buf + sizeof(*msg), hash, POA_QUERY_HLEN); + + if (poa->ops->poa_send_mgmt(poa, src, buf, sizeof(buf)) < 0) + return; /* the requester retransmits */ + + POA_STAT_BUMP(poa, rep_tx); +} + +static bool mac_is_zero(const uint8_t * mac) +{ + static const uint8_t zero[POA_MAC_SIZE] = { 0 }; + + return memcmp(mac, zero, POA_MAC_SIZE) == 0; +} + +static bool eth_query_is_match(const struct eth_query * q, + const struct poa * poa) +{ + if (q->ethertype != 0 && + q->ethertype != poa->local.eth.src.ethertype) + return false; + + if (q->c_dev[0] != '\0' && + strcmp(q->c_dev, poa->local.eth.src.dev) != 0) + return false; + + if (mac_is_zero(q->c_mac)) + return true; + + return memcmp(q->c_mac, poa->local.eth.src.mac, POA_MAC_SIZE) == 0; +} + +static void eth_reply_arr(struct poa * poa, + const struct poa_addr * src, + const uint8_t * hash) +{ + struct list_head * p; + + pthread_once(&queries.once, queries_init); + + pthread_mutex_lock(&queries.mtx); + + list_for_each(p, &queries.pending.list) { + struct eth_query * q; + + q = list_entry(p, struct eth_query, next); + if (q->replied || memcmp(q->hash, hash, POA_QUERY_HLEN) != 0) + continue; + + if (!eth_query_is_match(q, poa)) + continue; + + memcpy(q->mac, src->eth.dst.mac, POA_MAC_SIZE); + memcpy(q->src_mac, src->eth.src.mac, POA_MAC_SIZE); + strcpy(q->dev, src->eth.src.dev); + + q->r_ethertype = poa->local.eth.src.ethertype; + + q->replied = true; + } + + pthread_cond_broadcast(&queries.cond); + + pthread_mutex_unlock(&queries.mtx); +} + +/* Name query frames are handled in the transport. */ +static void eth_rx_mgmt(struct poa * poa, + const struct poa_addr * src, + const uint8_t * buf, + size_t len) +{ + const struct poa_mgmt_msg * msg; + const uint8_t * hash; + + msg = (const struct poa_mgmt_msg *) buf; + if (len < sizeof(*msg)) { + poa_rx_mgmt(poa, src, buf, len); + return; + } + + if (msg->code != POA_NAME_QUERY && msg->code != POA_NAME_REPLY) { + poa_rx_mgmt(poa, src, buf, len); + return; + } + + if (ntoh16(msg->data_len) != POA_QUERY_HLEN) + return; /* malformed */ + + if (len < sizeof(*msg) + POA_QUERY_HLEN) + return; /* malformed */ + + hash = buf + sizeof(*msg); + + if (msg->code == POA_NAME_QUERY) { + POA_STAT_BUMP(poa, qry_rx); + eth_query_arr(poa, src, hash); + } else { + POA_STAT_BUMP(poa, rep_rx); + eth_reply_arr(poa, src, hash); + } +} + +static int eth_query_send(const uint8_t * hash, + const struct poa_addr * addr) +{ + uint8_t buf[sizeof(struct poa_mgmt_msg) + + POA_QUERY_HLEN]; + struct poa_mgmt_msg * msg = (struct poa_mgmt_msg *) buf; + struct poa_addr bcast; + + bcast = *addr; + + memset(bcast.eth.dst.mac, 0xff, POA_MAC_SIZE); + + poa_mgmt_msg_ser(msg, POA_NAME_QUERY, 0, 0, qos_raw, 0, POA_QUERY_HLEN); + + memcpy(buf + sizeof(*msg), hash, POA_QUERY_HLEN); + + return poa_bcast_mgmt(&bcast, buf, sizeof(buf)); +} + +/* Complete addr for dst by broadcast query; the poa_query op. */ +static int eth_query(const char * dst, + const struct timespec * timeo, + struct poa_addr * addr) +{ + struct timespec dflt = TIMESPEC_INIT_MS(ETH_QUERY_TIMEO); + struct timespec rintv = TIMESPEC_INIT_MS(ETH_QUERY_TIMEO + / (ETH_QUERY_RETRIES + 1)); + struct eth_query q; + struct timespec abstime; + struct timespec now; + struct timespec dl; + time_t rintv_ns = (time_t) TS_TO_UINT64(rintv); + time_t diff; + uint8_t hash[POA_QUERY_HLEN]; + int n; + int err = -ETIMEDOUT; + + if (strlen(addr->eth.src.dev) > DEV_NAME_SIZE) + return -EINVAL; + + addr->eth.src.ethertype = addr->eth.dst.ethertype; + + /* The destination is set; a zero ethertype cannot be sent. */ + if (!mac_is_zero(addr->eth.dst.mac)) + return addr->eth.dst.ethertype < 0x0600 ? -EINVAL : 0; + + str_hash(HASH_SHA3_256, hash, dst); + + pthread_once(&queries.once, queries_init); + + memset(&q, 0, sizeof(q)); + memcpy(q.hash, hash, POA_QUERY_HLEN); + + q.ethertype = addr->eth.dst.ethertype; + + memcpy(q.c_mac, addr->eth.src.mac, POA_MAC_SIZE); + strcpy(q.c_dev, addr->eth.src.dev); + + pthread_mutex_lock(&queries.mtx); + llist_add(&q.next, &queries.pending); + pthread_mutex_unlock(&queries.mtx); + + pthread_cleanup_push(eth_query_cleanup, &q); + + clock_gettime(PTHREAD_COND_CLOCK, &now); + ts_add(&now, timeo != NULL ? timeo : &dflt, &abstime); + + while (ts_diff_ns(&now, &abstime) < 0) { + n = eth_query_send(hash, addr); + if (n < 0) { + err = n; + break; + } + + if (n == 0) { + err = -EPERM; + break; + } + + ts_add(&now, &rintv, &dl); + + if (ts_diff_ns(&dl, &abstime) > 0) + dl = abstime; + + pthread_mutex_lock(&queries.mtx); + + pthread_cleanup_push(__cleanup_mutex_unlock, &queries.mtx); + + while (!q.replied) { + if (pthread_cond_timedwait(&queries.cond, &queries.mtx, + &dl) == ETIMEDOUT) + break; + } + + if (q.replied) { + memcpy(addr->eth.dst.mac, q.mac, POA_MAC_SIZE); + memcpy(addr->eth.src.mac, q.src_mac, POA_MAC_SIZE); + strcpy(addr->eth.src.dev, q.dev); + + addr->eth.dst.ethertype = q.r_ethertype; + addr->eth.src.ethertype = q.r_ethertype; + + err = 0; + } + + pthread_cleanup_pop(true); + + if (err == 0) + break; + + clock_gettime(PTHREAD_COND_CLOCK, &now); + + diff = ts_diff_ns(&now, &abstime); + if (diff > -rintv_ns) /* skip the runt attempt */ + break; + } + + pthread_cleanup_pop(true); + + return err; +} + +static void eth_spec(const struct poa * poa, + struct poa_spec * spec) +{ + spec->type = poa->type; + spec->eth = poa->local.eth.src; +} + +static bool eth_has_id(const struct poa * poa, + const struct poa_spec * spec) +{ + if (strnlen(spec->eth.dev, sizeof(spec->eth.dev)) > DEV_NAME_SIZE) + return false; + + if (strcmp(poa->local.eth.src.dev, spec->eth.dev) != 0) + return false; + + return poa->local.eth.src.ethertype == spec->eth.ethertype; +} + +/* Our end of the link; an unnamed one matches any. */ +static bool eth_is_src(const struct poa * poa, + const struct eth_poa * src) +{ + if (src->dev[0] != '\0' && + strcmp(poa->local.eth.src.dev, src->dev) != 0) + return false; + + if (mac_is_zero(src->mac)) + return true; + + return memcmp(poa->local.eth.src.mac, src->mac, + POA_MAC_SIZE) == 0; +} + +/* An ethertype of 0 in dst matches any, for broadcast queries. */ +static bool eth_match(const struct poa * poa, + const struct poa_addr * dst) +{ + uint16_t et = dst->eth.dst.ethertype; + + if (et != 0 && et != poa->local.eth.src.ethertype) + return false; + + return eth_is_src(poa, &dst->eth.src); +} + +static bool eth_link_match(const struct poa * poa, + int id) +{ + struct eth_priv * priv = (struct eth_priv *) poa->priv; + + return priv->ifindex == id; +} + +#ifdef HAVE_RAW_SOCKETS + +/* MSG_DONTWAIT: the reader blocks on this socket. */ +static int eth_sendv(struct eth_priv * priv, + const uint8_t * dst, + uint32_t eid, + const uint8_t * body, + size_t len, + bool block, + const struct timespec * abstime) +{ + struct eth_hdr hdr; + struct msghdr msg; + struct iovec iov[2]; + int ret; + + if (len > priv->mtu - POA_HDR_SIZE) + return -EMSGSIZE; + + eth_hdr_ser(priv, &hdr, dst, eid, len); + + iov[0].iov_base = &hdr; + iov[0].iov_len = ETH_HDR_TOT_SIZE; + iov[1].iov_base = (void *) body; + iov[1].iov_len = len; + + memset(&msg, 0, sizeof(msg)); + + msg.msg_name = &priv->device; + msg.msg_namelen = sizeof(priv->device); + msg.msg_iov = iov; + msg.msg_iovlen = len > 0 ? 2 : 1; + while (sendmsg(priv->s_fd, &msg, MSG_DONTWAIT) < 0) { + if (errno != EAGAIN && errno != EWOULDBLOCK) + return -EIO; + + if (!block) + return -EAGAIN; + + ret = poa_wait_out(priv->s_fd, abstime); + if (ret < 0) + return ret; + } + + return 0; +} + +static void * eth_reader(void * o) +{ + struct poa * poa = (struct poa *) o; + struct eth_priv * priv = (struct eth_priv *) poa->priv; + struct timespec down = TIMESPEC_INIT_MS(ETH_DOWN_TIMEO); + uint8_t * buf; + size_t bufsz; + + bufsz = ETH_HDR_SIZE + priv->mtu; + + buf = malloc(bufsz); + if (buf == NULL) + return (void *) -1; + + pthread_cleanup_push(free, buf); + + while (true) { + struct ssm_pk_buff * spb; + struct sockaddr_ll from; + socklen_t flen; + struct poa_addr src; + ssize_t n; + uint32_t eid; + size_t plen; + const uint8_t * body; + + flen = sizeof(from); + + n = recvfrom(priv->s_fd, buf, bufsz, 0, + (struct sockaddr *) &from, &flen); + + if (n < 0) { + if (errno == EINTR) + continue; + + POA_STAT_BUMP(poa, rcv_fail); + + if (errno == ENETDOWN) { + nanosleep(&down, NULL); + continue; + } + + log_err("Reader on %s stopped: %s.", + poa->local.eth.src.dev, + strerror(errno)); + break; + } + + if (from.sll_pkttype == PACKET_OUTGOING) + continue; + + if (frame_parse(priv, buf, (size_t) n, &eid, &plen) < 0) + continue; + + body = buf + ETH_HDR_TOT_SIZE; + + if (eid == POA_MGMT_EID) { + frame_to_addr(priv, (struct eth_hdr *) buf, &src); + eth_rx_mgmt(poa, &src, body, plen); + continue; + } + + if (poa_spb_reserve(&spb, plen) < 0) { + POA_STAT_BUMP(poa, buf_fail); + continue; + } + + memcpy(ssm_pk_buff_head(spb), body, plen); + + poa_rx_pkt(poa, eid, spb); + } + + pthread_cleanup_pop(true); + + return (void *) 0; +} + +/* + * One netlink socket for the whole subsystem: RTMGRP_LINK delivers the + * events of every interface anyway, so a socket per PoA only added + * discards. + */ +int poa_monitor_open(void) +{ + struct sockaddr_nl sa; + int fd; + + memset(&sa, 0, sizeof(sa)); + + sa.nl_family = AF_NETLINK; + sa.nl_groups = RTMGRP_LINK; + + fd = socket(AF_NETLINK, SOCK_RAW, NETLINK_ROUTE); + if (fd < 0) + return -1; + + if (bind(fd, (struct sockaddr *) &sa, sizeof(sa)) < 0) { + close(fd); + return -1; + } + + return fd; +} + +void poa_monitor_read(int fd) +{ + uint8_t buf[4096]; + struct nlmsghdr * h; + ssize_t n; + + n = recv(fd, buf, sizeof(buf), 0); + if (n < 0) + return; + + for (h = (struct nlmsghdr *) buf; + NLMSG_OK(h, (unsigned int) n); + h = NLMSG_NEXT(h, n)) { + struct ifinfomsg * ifi; + unsigned int usable; + bool up; + size_t cnt; + + if (h->nlmsg_type == NLMSG_DONE) + break; + + if (h->nlmsg_type != RTM_NEWLINK) + continue; + + ifi = NLMSG_DATA(h); + + usable = ifi->ifi_flags & (IFF_UP | IFF_RUNNING); + up = usable == (IFF_UP | IFF_RUNNING); + + cnt = poa_link_updown(ifi->ifi_index, up); + if (cnt > 0) + log_info("Link %d %s, %zu flows.", ifi->ifi_index, + up ? "up" : "down", cnt); + } +} + +static int eth_dev_info(const char * dev, + int * idx, + uint8_t * mac, + uint32_t * mtu) +{ + struct ifreq ifr; + int fd; + + if (strlen(dev) >= IFNAMSIZ) + return -EINVAL; + + *idx = if_nametoindex(dev); + if (*idx == 0) { + log_err("Failed to find device %s.", dev); + return -ENODEV; + } + + fd = socket(AF_UNIX, SOCK_STREAM, 0); + if (fd < 0) + return -EIO; + + memset(&ifr, 0, sizeof(ifr)); + strcpy(ifr.ifr_name, dev); + + if (ioctl(fd, SIOCGIFHWADDR, &ifr) < 0) { + log_err("Failed to get hardware address of %s.", dev); + goto fail_ioctl; + } + + memcpy(mac, ifr.ifr_hwaddr.sa_data, POA_MAC_SIZE); + + if (ioctl(fd, SIOCGIFMTU, &ifr) < 0) { + log_err("Failed to get MTU of %s.", dev); + goto fail_ioctl; + } + + close(fd); + + *mtu = MIN(MIN(ETH_MAX_MTU, POA_ETH_RD_BUF), (uint32_t) ifr.ifr_mtu); + if (memcmp(dev, "lo", 2) == 0 && *mtu > POA_ETH_LO_MTU) + *mtu = POA_ETH_LO_MTU; + + return 0; + + fail_ioctl: + close(fd); + return -EIO; +} + +/* SO_RCVBUFFORCE bypasses rmem_max; SO_RCVBUF is the fallback. */ +static void eth_set_rcvbuf(int fd, + int rcvbuf) +{ + if (setsockopt(fd, SOL_SOCKET, SO_RCVBUFFORCE, + &rcvbuf, sizeof(rcvbuf)) == 0) + return; + + if (setsockopt(fd, SOL_SOCKET, SO_RCVBUF, &rcvbuf, sizeof(rcvbuf)) < 0) + log_info("Failed to set SO_RCVBUF to %d.", rcvbuf); +} + +/* The send buffer holds at least one frame, or sendmsg cannot fit a PDU. */ +static int eth_socket(struct eth_priv * priv, + int idx, + const uint8_t * mac) +{ + int sndbuf; + int rcvbuf; +#ifdef POA_ETH_QDISC_BYPASS + int bypass = 1; +#endif + + memset(&priv->device, 0, sizeof(priv->device)); + + priv->device.sll_ifindex = idx; + priv->device.sll_family = AF_PACKET; + priv->device.sll_halen = POA_MAC_SIZE; + priv->device.sll_protocol = htons(ETH_P_ALL); + + memcpy(priv->device.sll_addr, mac, POA_MAC_SIZE); + memcpy(priv->hw_addr, mac, POA_MAC_SIZE); + + priv->s_fd = socket(AF_PACKET, SOCK_RAW, priv->ethertype); + if (priv->s_fd < 0) { + log_err("Failed to create socket: %s.", strerror(errno)); + return -1; + } + +#ifdef POA_ETH_QDISC_BYPASS + if (setsockopt(priv->s_fd, SOL_PACKET, PACKET_QDISC_BYPASS, + &bypass, sizeof(bypass)) < 0) + log_info("Qdisc bypass not supported."); +#endif + + sndbuf = POA_ETH_SNDBUF; + if (sndbuf > 0) { + sndbuf = MAX(sndbuf, (int) (ETH_HDR_SIZE + priv->mtu)); + + if (setsockopt(priv->s_fd, SOL_SOCKET, SO_SNDBUF, + &sndbuf, sizeof(sndbuf)) < 0) + log_info("Failed to set SO_SNDBUF to %d.", sndbuf); + } + + rcvbuf = POA_ETH_RCVBUF; + if (rcvbuf > 0) + eth_set_rcvbuf(priv->s_fd, rcvbuf); + + if (bind(priv->s_fd, (struct sockaddr *) &priv->device, + sizeof(priv->device)) < 0) { + log_err("Failed to bind socket to %d.", idx); + close(priv->s_fd); + return -1; + } + + return 0; +} + +/* + * Pull qlen and backlog from the nested TCA_STATS2. The top-level + * TCA_STATS shares TCA_STATS_QUEUE's id but carries a wholly + * different struct, so descend first; never match on id alone. + */ +static void eth_qdisc_parse(struct rtattr * rta, + size_t rlen, + size_t * byt, + size_t * pkt) +{ + struct gnet_stats_queue q; + struct rtattr * in; + size_t ilen; + + for (; RTA_OK(rta, rlen); rta = RTA_NEXT(rta, rlen)) { + if (rta->rta_type != TCA_STATS2) + continue; + + in = (struct rtattr *) RTA_DATA(rta); + ilen = RTA_PAYLOAD(rta); + for (; RTA_OK(in, ilen); in = RTA_NEXT(in, ilen)) { + if (in->rta_type != TCA_STATS_QUEUE) + continue; + + if (RTA_PAYLOAD(in) < sizeof(q)) + continue; + + memcpy(&q, RTA_DATA(in), sizeof(q)); + + *byt = q.backlog; + *pkt = q.qlen; + } + } +} + +/* + * Egress backlog of the device's root qdisc, in bytes and packets, + * as the kernel queues them. The caller owns the netlink descriptor + * and serialises the query. An early end of dump reports failure. + */ +static int eth_qdisc_backlog(int fd, + int ifindex, + size_t * byt, + size_t * pkt) +{ + struct { + struct nlmsghdr nh; + struct tcmsg tc; + } req; + struct nlmsghdr * nh; + struct rtattr * rta; + struct tcmsg * tc; + char buf[16384]; + ssize_t len; + int ret = -1; + + if (fd < 0) + goto fail; + + memset(&req, 0, sizeof(req)); + + req.nh.nlmsg_len = NLMSG_LENGTH(sizeof(req.tc)); + req.nh.nlmsg_type = RTM_GETQDISC; + req.nh.nlmsg_flags = NLM_F_REQUEST | NLM_F_DUMP; + req.nh.nlmsg_seq = 1; + req.tc.tcm_family = AF_UNSPEC; + req.tc.tcm_ifindex = ifindex; + if (send(fd, &req, req.nh.nlmsg_len, 0) < 0) + goto fail; + + *byt = 0; + *pkt = 0; + + while ((len = recv(fd, buf, sizeof(buf), 0)) > 0) { + nh = (struct nlmsghdr *) buf; + for (; NLMSG_OK(nh, len); nh = NLMSG_NEXT(nh, len)) { + if (nh->nlmsg_type == NLMSG_DONE) + goto done; + + if (nh->nlmsg_type == NLMSG_ERROR) + goto fail; + + if (nh->nlmsg_type != RTM_NEWQDISC) + continue; + + tc = (struct tcmsg *) NLMSG_DATA(nh); + if (tc->tcm_ifindex != ifindex) + continue; + + if (tc->tcm_parent != TC_H_ROOT) + continue; + + rta = (struct rtattr *) + ((char *) tc + NLMSG_ALIGN(sizeof(*tc))); + + eth_qdisc_parse(rta, nh->nlmsg_len + - NLMSG_LENGTH(sizeof(*tc)), + byt, pkt); + } + } + + goto fail; /* early end: a zero would read as empty */ + done: + ret = 0; + fail: + return ret; +} + +/* + * A netlink descriptor for the qdisc query. SO_RCVTIMEO bounds the + * dump: this is read from the send path, and a reply that never + * arrives must not park a sender thread. + */ +static int eth_qdisc_open(void) +{ + struct sockaddr_nl sa; + struct timeval tv = TIMEVAL_INIT_MS(POA_QDISC_TIMEO); + int fd; + + fd = socket(AF_NETLINK, SOCK_RAW | SOCK_CLOEXEC, NETLINK_ROUTE); + if (fd < 0) + return -1; + + memset(&sa, 0, sizeof(sa)); + + sa.nl_family = AF_NETLINK; + if (bind(fd, (struct sockaddr *) &sa, sizeof(sa)) < 0) + goto fail; + + if (setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv)) < 0) + goto fail; + + return fd; + fail: + close(fd); + + return -1; +} + +/* + * Queue depth in packets, straight from the qdisc. Gated like the + * transport's own depth and skipped when another sender is already + * asking, so the send path never waits on netlink. + */ +static int eth_qpkts(struct poa * poa, + size_t * pkts, + size_t * byts) +{ + struct eth_priv * priv = (struct eth_priv *) poa->priv; + struct timespec now; + uint64_t ns; + size_t byt; + size_t pkt; + + if (priv->nl_fd < 0) + return -1; + + clock_gettime(PTHREAD_COND_CLOCK, &now); + + ns = TS_TO_UINT64(now); + if (ns - LOAD_RELAXED(&priv->nl_time) < POA_QLEN_GATE) + goto cached; + + if (__atomic_exchange_n(&priv->nl_busy, 1, __ATOMIC_ACQUIRE) != 0) + goto cached; + + if (eth_qdisc_backlog(priv->nl_fd, priv->ifindex, &byt, &pkt) == 0) { + STORE_RELAXED(&priv->nl_pkt, pkt); + STORE_RELAXED(&priv->nl_byt, byt); + STORE_RELAXED(&priv->nl_time, ns); + } + + __atomic_store_n(&priv->nl_busy, 0, __ATOMIC_RELEASE); + cached: + if (LOAD_RELAXED(&priv->nl_time) == 0) /* nothing measured yet */ + return -1; + + *pkts = LOAD_RELAXED(&priv->nl_pkt); + *byts = LOAD_RELAXED(&priv->nl_byt); + + return 0; +} + +static int eth_attach(struct poa * poa, + const struct poa_spec * spec) +{ + const struct eth_poa * c = &spec->eth; + struct eth_priv * priv; + uint8_t mac[POA_MAC_SIZE]; + uint32_t mtu; + int idx; + int err; + + if (strnlen(c->dev, sizeof(c->dev)) > DEV_NAME_SIZE) + return -EINVAL; + + if (c->ethertype < 0x0600 || c->ethertype == 0xFFFF) { + log_err("Invalid Ethertype 0x%x.", c->ethertype); + return -EINVAL; + } + + priv = malloc(sizeof(*priv)); + if (priv == NULL) + return -ENOMEM; + + memset(priv, 0, sizeof(*priv)); + + priv->poa = poa; + priv->nl_fd = eth_qdisc_open(); /* optional; socket fallback */ + + priv->ethertype = htons(c->ethertype); + + err = eth_dev_info(c->dev, &idx, mac, &mtu); + if (err < 0) + goto fail_conf; + + priv->mtu = mtu; + if (eth_socket(priv, idx, mac) < 0) + goto fail_conf; + + poa->priv = priv; + poa->local.type = poa->type; + poa->local.eth.src.ethertype = c->ethertype; + priv->ifindex = idx; + + memcpy(poa->local.eth.src.mac, mac, POA_MAC_SIZE); + strcpy(poa->local.eth.src.dev, c->dev); + + return 0; + + fail_conf: + if (priv->nl_fd >= 0) + close(priv->nl_fd); + + free(priv); + + return -EIO; +} + +static void eth_detach(struct poa * poa) +{ + struct eth_priv * priv = (struct eth_priv *) poa->priv; + + if (priv == NULL) + return; + + close(priv->s_fd); + + if (priv->nl_fd >= 0) + close(priv->nl_fd); + + free(priv); + + poa->priv = NULL; +} + +static uint32_t eth_mtu(struct poa * poa, + const struct poa_addr * dst) +{ + struct eth_priv * priv = (struct eth_priv *) poa->priv; + + (void) dst; + + return priv->mtu - POA_HDR_SIZE; +} + +/* All flows on the PoA share the socket, so this is aggregate. */ +static size_t eth_qlen(struct poa * poa) +{ +#ifdef SIOCOUTQ + struct eth_priv * priv = (struct eth_priv *) poa->priv; + int qlen; + + qlen = 0; + if (ioctl(priv->s_fd, SIOCOUTQ, &qlen) < 0) + return 0; + + return (size_t) qlen; +#else + (void) poa; + + return 0; +#endif +} + +/* + * PACKET_STATISTICS zeroes the kernel counters on read, so totals + * accumulate here; relaxed atomics allow concurrent RIB reads. + * The qdisc depth rides the sender's guarded path: one dump at a time. + */ +static int eth_rib(struct poa * poa, + char * buf, + size_t len) +{ + struct eth_priv * priv = (struct eth_priv *) poa->priv; + struct tpacket_stats ts; + socklen_t optlen; + size_t sndbuf = 0; + size_t rcvbuf = 0; + size_t qd_byt; + size_t qd_pkt; + int val; + int size; + + optlen = sizeof(val); + if (getsockopt(priv->s_fd, SOL_SOCKET, SO_SNDBUF, &val, &optlen) == 0) + sndbuf = (size_t) val; + + optlen = sizeof(val); + if (getsockopt(priv->s_fd, SOL_SOCKET, SO_RCVBUF, &val, &optlen) == 0) + rcvbuf = (size_t) val; + + optlen = sizeof(ts); + + if (getsockopt(priv->s_fd, SOL_PACKET, PACKET_STATISTICS, + &ts, &optlen) == 0) { + FETCH_ADD_RELAXED(&priv->kern_rcv, ts.tp_packets); + FETCH_ADD_RELAXED(&priv->kern_drp, ts.tp_drops); + } + + if (eth_qpkts(poa, &qd_pkt, &qd_byt) < 0) { + qd_byt = 0; + qd_pkt = 0; + } + + size = snprintf(buf, len, + "Socket sndbuf (bytes): %zu\n" + "Socket rcvbuf (bytes): %zu\n" + "Kernel packets received: %zu\n" + "Kernel packets dropped: %zu\n" + "Qdisc backlog (bytes): %zu\n" + "Qdisc backlog (packets): %zu\n", + sndbuf, rcvbuf, + LOAD_RELAXED(&priv->kern_rcv), + LOAD_RELAXED(&priv->kern_drp), + qd_byt, qd_pkt); + if (size < 0 || (size_t) size >= len) + return -1; + + return size; +} + +#elif defined(HAVE_BPF) + +/* + * BSD and macOS reach the link layer through a cloned /dev/bpf + * device, bound to an interface with BIOCSETIF. One device per PoA. + */ + +static int eth_sendv(struct eth_priv * priv, + const uint8_t * dst, + uint32_t eid, + const uint8_t * body, + size_t len, + bool block, + const struct timespec * abstime) +{ + struct eth_hdr hdr; + struct iovec iov[2]; + int n; + int ret; + + if (len > priv->mtu - POA_HDR_SIZE) + return -EMSGSIZE; + + eth_hdr_ser(priv, &hdr, dst, eid, len); + + iov[0].iov_base = &hdr; + iov[0].iov_len = ETH_HDR_TOT_SIZE; + iov[1].iov_base = (void *) body; + iov[1].iov_len = len; + n = len > 0 ? 2 : 1; + while (writev(priv->s_fd, iov, n) < 0) { + if (errno != EAGAIN && errno != EWOULDBLOCK) + return -EIO; + + if (!block) + return -EAGAIN; + + ret = poa_wait_out(priv->s_fd, abstime); + if (ret < 0) + return ret; + } + + return 0; +} + +/* One read yields a batch of BPF_WORDALIGN'ed frames; walk all of them. */ +static void * eth_reader(void * o) +{ + struct poa * poa = (struct poa *) o; + struct eth_priv * priv = (struct eth_priv *) poa->priv; + uint8_t * buf; + + buf = malloc(priv->blen); + if (buf == NULL) + return (void *) -1; + + pthread_cleanup_push(free, buf); + + while (true) { + uint8_t * p; + uint8_t * end; + ssize_t n; + + n = read(priv->s_fd, buf, priv->blen); + if (n < 0) { + if (errno == EINTR) + continue; + + POA_STAT_BUMP(poa, rcv_fail); + break; + } + + p = buf; + end = buf + n; + while (p + sizeof(struct bpf_hdr) <= end) { + struct bpf_hdr * bh = (struct bpf_hdr *) p; + struct ssm_pk_buff * spb; + struct poa_addr src; + const uint8_t * frame; + const uint8_t * body; + uint32_t eid; + size_t plen; + + frame = p + bh->bh_hdrlen; + if (frame + bh->bh_caplen > end) + break; + + if (frame_parse(priv, frame, bh->bh_caplen, + &eid, &plen) < 0) + goto next; + + body = frame + ETH_HDR_TOT_SIZE; + + if (eid == POA_MGMT_EID) { + frame_to_addr(priv, + (const struct eth_hdr *) frame, + &src); + eth_rx_mgmt(poa, &src, body, plen); + goto next; + } + + if (poa_spb_reserve(&spb, plen) < 0) { + POA_STAT_BUMP(poa, buf_fail); + goto next; + } + + memcpy(ssm_pk_buff_head(spb), body, plen); + + poa_rx_pkt(poa, eid, spb); + next: + p += BPF_WORDALIGN(bh->bh_hdrlen + bh->bh_caplen); + } + } + + pthread_cleanup_pop(true); + + return (void *) 0; +} + +static int eth_dev_info(const char * dev, + uint8_t * mac, + uint32_t * mtu) +{ + struct ifaddrs * ifas; + struct ifaddrs * ifa; + struct ifreq ifr; + int fd; + int found = 0; + + if (getifaddrs(&ifas) < 0) { + log_err("Failed to list interfaces."); + return -1; + } + + for (ifa = ifas; ifa != NULL; ifa = ifa->ifa_next) { + struct sockaddr_dl * dl; + + if (ifa->ifa_addr == NULL) + continue; + + if (ifa->ifa_addr->sa_family != AF_LINK) + continue; + + if (strcmp(ifa->ifa_name, dev) != 0) + continue; + + dl = (struct sockaddr_dl *) ifa->ifa_addr; + if (dl->sdl_alen != POA_MAC_SIZE) + continue; + + memcpy(mac, LLADDR(dl), POA_MAC_SIZE); + + found = 1; + break; + } + + freeifaddrs(ifas); + + if (!found) { + log_err("No such device: %s.", dev); + return -1; + } + + fd = socket(AF_INET, SOCK_DGRAM, 0); + if (fd < 0) + return -1; + + memset(&ifr, 0, sizeof(ifr)); + + strcpy(ifr.ifr_name, dev); + + if (ioctl(fd, SIOCGIFMTU, &ifr) < 0) { + log_err("Failed to get MTU of %s.", dev); + close(fd); + return -1; + } + + close(fd); + + *mtu = MIN(MIN(ETH_MAX_MTU, POA_ETH_RD_BUF), (uint32_t) ifr.ifr_mtu); + if (memcmp(dev, "lo", 2) == 0 && *mtu > POA_ETH_LO_MTU) + *mtu = POA_ETH_LO_MTU; + + return 0; +} + +static int eth_bpf_open(void) +{ + char dev[32]; + size_t i; + + for (i = 0; i < BPF_DEV_MAX; ++i) { + int fd; + + sprintf(dev, "/dev/bpf%zu", i); + + fd = open(dev, O_RDWR); + if (fd >= 0) + return fd; + } + + return -1; +} + +/* + * BIOCSHDRCMPLT: we fill in the source address ourselves. + * BIOCSSEESENT: our own egress must not come back at us. + * BIOCIMMEDIATE: deliver on arrival, do not wait for a full buffer. + */ +static int eth_bpf_setup(struct eth_priv * priv, + const char * dev) +{ + struct ifreq ifr; + int enable = 1; + int disable = 0; + int blen = 0; + + memset(&ifr, 0, sizeof(ifr)); + + strcpy(ifr.ifr_name, dev); + + if (ioctl(priv->s_fd, BIOCSETIF, &ifr) < 0) { + log_err("Failed to bind bpf device to %s.", dev); + return -1; + } + + if (ioctl(priv->s_fd, BIOCGBLEN, &blen) < 0 || blen <= 0) { + log_err("Failed to get the bpf buffer length."); + return -1; + } + + priv->blen = (size_t) blen; + if (ioctl(priv->s_fd, BIOCSHDRCMPLT, &enable) < 0) { + log_err("Failed to set BIOCSHDRCMPLT."); + return -1; + } + + if (ioctl(priv->s_fd, BIOCSSEESENT, &disable) < 0) { + log_err("Failed to set BIOCSSEESENT."); + return -1; + } + + if (ioctl(priv->s_fd, BIOCIMMEDIATE, &enable) < 0) { + log_err("Failed to set BIOCIMMEDIATE."); + return -1; + } + + return 0; +} + +static int eth_attach(struct poa * poa, + const struct poa_spec * spec) +{ + const struct eth_poa * c = &spec->eth; + struct eth_priv * priv; + uint8_t mac[POA_MAC_SIZE]; + uint32_t mtu; + + if (strnlen(c->dev, sizeof(c->dev)) > DEV_NAME_SIZE) + return -EINVAL; + + if (c->ethertype < 0x0600 || c->ethertype == 0xFFFF) { + log_err("Invalid Ethertype 0x%x.", c->ethertype); + return -EINVAL; + } + + priv = malloc(sizeof(*priv)); + if (priv == NULL) + return -ENOMEM; + + memset(priv, 0, sizeof(*priv)); + + priv->poa = poa; + priv->s_fd = -1; + + priv->ethertype = htons(c->ethertype); + + if (eth_dev_info(c->dev, mac, &mtu) < 0) + goto fail_conf; + + priv->mtu = mtu; + + memcpy(priv->hw_addr, mac, POA_MAC_SIZE); + + priv->s_fd = eth_bpf_open(); + if (priv->s_fd < 0) { + log_err("Failed to open a bpf device."); + goto fail_conf; + } + + if (eth_bpf_setup(priv, c->dev) < 0) + goto fail_device; + + poa->priv = priv; + poa->local.type = poa->type; + poa->local.eth.src.ethertype = c->ethertype; + + memcpy(poa->local.eth.src.mac, mac, POA_MAC_SIZE); + strcpy(poa->local.eth.src.dev, c->dev); + + log_info("Using Berkeley Packet Filter on %s.", c->dev); + + return 0; + + fail_device: + close(priv->s_fd); + fail_conf: + free(priv); + + return -EIO; +} + +static void eth_detach(struct poa * poa) +{ + struct eth_priv * priv = (struct eth_priv *) poa->priv; + + if (priv == NULL) + return; + + close(priv->s_fd); + + free(priv); + + poa->priv = NULL; +} + +static uint32_t eth_mtu(struct poa * poa, + const struct poa_addr * dst) +{ + struct eth_priv * priv = (struct eth_priv *) poa->priv; + + (void) dst; + + return priv->mtu - POA_HDR_SIZE; +} + +/* A bpf device has no send queue to report; mb-ECN cannot mark here. */ +static size_t eth_qlen(struct poa * poa) +{ + (void) poa; + + return 0; +} + +/* The bpf device buffer is all the receive queue there is. */ +static int eth_rib(struct poa * poa, + char * buf, + size_t len) +{ + struct eth_priv * priv = (struct eth_priv *) poa->priv; + struct bpf_stat bs; + int size; + + if (ioctl(priv->s_fd, BIOCGSTATS, &bs) == 0) { + FETCH_ADD_RELAXED(&priv->kern_rcv, bs.bs_recv); + FETCH_ADD_RELAXED(&priv->kern_drp, bs.bs_drop); + } + + size = snprintf(buf, len, + "Socket rcvbuf (bytes): %zu\n" + "Kernel packets received: %zu\n" + "Kernel packets dropped: %zu\n", + priv->blen, + LOAD_RELAXED(&priv->kern_rcv), + LOAD_RELAXED(&priv->kern_drp)); + if (size < 0 || (size_t) size >= len) + return -1; + + return size; +} + +#elif defined(HAVE_NETMAP) + +/* + * netmap gives one PoA the whole port: reads copy out of the NIC + * ring, writes inject into it. + */ + +/* nm_inject takes one contiguous frame, so the header is copied in. */ +static int eth_sendv(struct eth_priv * priv, + const uint8_t * dst, + uint32_t eid, + const uint8_t * body, + size_t len, + bool block, + const struct timespec * abstime) +{ + uint8_t * frame; + size_t flen; + int ret; + + if (len > priv->mtu - POA_HDR_SIZE) + return -EMSGSIZE; + + flen = ETH_HDR_TOT_SIZE + len; + + frame = malloc(flen); + if (frame == NULL) + return -ENOMEM; + + eth_hdr_ser(priv, (struct eth_hdr *) frame, dst, eid, len); + + if (len > 0) + memcpy(frame + ETH_HDR_TOT_SIZE, body, len); + + if (block) + ret = poa_wait_out(priv->poll_out.fd, abstime); + else + ret = poll(&priv->poll_out, 1, 0) > 0 ? 0 : -EAGAIN; + + if (ret < 0) + goto fail; + + ret = nm_inject(priv->nmd, frame, flen) == (int) flen ? 0 : -EIO; + fail: + free(frame); + + return ret; +} + +/* A slot stays owned by the ring, so each frame is copied out. */ +static void * eth_reader(void * o) +{ + struct poa * poa = (struct poa *) o; + struct eth_priv * priv = (struct eth_priv *) poa->priv; + + while (true) { + struct ssm_pk_buff * spb; + struct poa_addr src; + struct nm_pkthdr hdr; + const uint8_t * frame; + const uint8_t * body; + uint32_t eid; + size_t plen; + + if (poll(&priv->poll_in, 1, -1) < 0) { + if (errno == EINTR) + continue; + + POA_STAT_BUMP(poa, rcv_fail); + break; + } + + if (priv->poll_in.revents == 0) + continue; + + frame = nm_nextpkt(priv->nmd, &hdr); + if (frame == NULL) + continue; + + if (frame_parse(priv, frame, hdr.len, &eid, &plen) < 0) + continue; + + body = frame + ETH_HDR_TOT_SIZE; + + if (eid == POA_MGMT_EID) { + frame_to_addr(priv, (const struct eth_hdr *) frame, + &src); + eth_rx_mgmt(poa, &src, body, plen); + continue; + } + + if (poa_spb_reserve(&spb, plen) < 0) { + POA_STAT_BUMP(poa, buf_fail); + continue; + } + + memcpy(ssm_pk_buff_head(spb), body, plen); + + poa_rx_pkt(poa, eid, spb); + } + + return (void *) 0; +} + +static int eth_dev_info(const char * dev, + uint8_t * mac, + uint32_t * mtu) +{ + struct ifreq ifr; +#ifndef __linux__ + struct ifaddrs * ifas; + struct ifaddrs * ifa; + int found = 0; +#endif + int fd; + + if (strlen(dev) >= IFNAMSIZ) + return -EINVAL; + + fd = socket(AF_INET, SOCK_DGRAM, 0); + if (fd < 0) + return -EIO; + + memset(&ifr, 0, sizeof(ifr)); + + strcpy(ifr.ifr_name, dev); + +#ifdef __linux__ + if (ioctl(fd, SIOCGIFHWADDR, &ifr) < 0) { + log_err("Failed to get hardware address of %s.", dev); + goto fail_ioctl; + } + + memcpy(mac, ifr.ifr_hwaddr.sa_data, POA_MAC_SIZE); +#else + if (getifaddrs(&ifas) < 0) + goto fail_ioctl; + + for (ifa = ifas; ifa != NULL; ifa = ifa->ifa_next) { + struct sockaddr_dl * dl; + + if (ifa->ifa_addr == NULL) + continue; + + if (ifa->ifa_addr->sa_family != AF_LINK) + continue; + + if (strcmp(ifa->ifa_name, dev) != 0) + continue; + + dl = (struct sockaddr_dl *) ifa->ifa_addr; + if (dl->sdl_alen != POA_MAC_SIZE) + continue; + + memcpy(mac, LLADDR(dl), POA_MAC_SIZE); + + found = 1; + break; + } + + freeifaddrs(ifas); + + if (!found) { + log_err("No hardware address for %s.", dev); + goto fail_ioctl; + } +#endif + if (ioctl(fd, SIOCGIFMTU, &ifr) < 0) { + log_err("Failed to get MTU of %s.", dev); + goto fail_ioctl; + } + + close(fd); + + *mtu = MIN(MIN(ETH_MAX_MTU, POA_ETH_RD_BUF), (uint32_t) ifr.ifr_mtu); + if (memcmp(dev, "lo", 2) == 0 && *mtu > POA_ETH_LO_MTU) + *mtu = POA_ETH_LO_MTU; + + return 0; + + fail_ioctl: + close(fd); + + return -EIO; +} + +static int eth_nm_open(struct eth_priv * priv, + const char * dev) +{ + char ifn[IFNAMSIZ + sizeof("netmap:")]; + + strcpy(ifn, "netmap:"); + strcat(ifn, dev); + + priv->nmd = nm_open(ifn, NULL, 0, NULL); + if (priv->nmd == NULL) { + log_err("Failed to open netmap device for %s.", dev); + return -1; + } + + memset(&priv->poll_in, 0, sizeof(priv->poll_in)); + memset(&priv->poll_out, 0, sizeof(priv->poll_out)); + + priv->poll_in.fd = NETMAP_FD(priv->nmd); + priv->poll_in.events = POLLIN; + priv->poll_out.fd = NETMAP_FD(priv->nmd); + priv->poll_out.events = POLLOUT; + + return 0; +} + +static int eth_attach(struct poa * poa, + const struct poa_spec * spec) +{ + const struct eth_poa * c = &spec->eth; + struct eth_priv * priv; + uint8_t mac[POA_MAC_SIZE]; + uint32_t mtu; + + if (strnlen(c->dev, sizeof(c->dev)) > DEV_NAME_SIZE) + return -EINVAL; + + if (c->ethertype < 0x0600 || c->ethertype == 0xFFFF) { + log_err("Invalid Ethertype 0x%x.", c->ethertype); + return -EINVAL; + } + + priv = malloc(sizeof(*priv)); + if (priv == NULL) + return -ENOMEM; + + memset(priv, 0, sizeof(*priv)); + + priv->poa = poa; + + priv->ethertype = htons(c->ethertype); + + if (eth_dev_info(c->dev, mac, &mtu) < 0) + goto fail_conf; + + priv->mtu = mtu; + + memcpy(priv->hw_addr, mac, POA_MAC_SIZE); + + if (eth_nm_open(priv, c->dev) < 0) + goto fail_conf; + + poa->priv = priv; + poa->local.type = poa->type; + poa->local.eth.src.ethertype = c->ethertype; + + memcpy(poa->local.eth.src.mac, mac, POA_MAC_SIZE); + strcpy(poa->local.eth.src.dev, c->dev); + + log_info("Using netmap on %s.", c->dev); + + return 0; + + fail_conf: + free(priv); + + return -EIO; +} + +static void eth_detach(struct poa * poa) +{ + struct eth_priv * priv = (struct eth_priv *) poa->priv; + + if (priv == NULL) + return; + + nm_close(priv->nmd); + + free(priv); + + poa->priv = NULL; +} + +static uint32_t eth_mtu(struct poa * poa, + const struct poa_addr * dst) +{ + struct eth_priv * priv = (struct eth_priv *) poa->priv; + + (void) dst; + + return priv->mtu - POA_HDR_SIZE; +} + +/* The ring is drained by the NIC; there is no queue to report. */ +static size_t eth_qlen(struct poa * poa) +{ + (void) poa; + + return 0; +} + +#endif /* HAVE_RAW_SOCKETS */ + +#ifndef HAVE_RAW_SOCKETS + +/* Only netlink reports link events; no other backend has a monitor. */ +int poa_monitor_open(void) +{ + return -1; +} + +void poa_monitor_read(int fd) +{ + (void) fd; +} + +#endif + +/* One reader per socket, so a flow cannot be reordered on receive. */ +static int eth_start(struct poa * poa) +{ + struct eth_priv * priv = (struct eth_priv *) poa->priv; + + if (pthread_create(&priv->reader, NULL, eth_reader, poa) != 0) + return -1; + + priv->running = true; + + return 0; +} + +static void eth_stop(struct poa * poa) +{ + struct eth_priv * priv = (struct eth_priv *) poa->priv; + + if (!priv->running) + return; + + pthread_cancel(priv->reader); + pthread_join(priv->reader, NULL); + + priv->running = false; +} + +static int eth_send(struct poa * poa, + const struct poa_addr * dst, + uint32_t eid, + struct ssm_pk_buff * spb, + bool block, + const struct timespec * abstime) +{ + return eth_sendv((struct eth_priv *) poa->priv, dst->eth.dst.mac, eid, + ssm_pk_buff_head(spb), ssm_pk_buff_len(spb), + block, abstime); +} + +static int eth_send_mgmt(struct poa * poa, + const struct poa_addr * dst, + const uint8_t * buf, + size_t len) +{ + struct timespec timeo = TIMESPEC_INIT_MS(POA_MGMT_SND_TIMEO); + struct timespec abstime; + + clock_gettime(PTHREAD_COND_CLOCK, &abstime); + ts_add(&abstime, &timeo, &abstime); + + return eth_sendv((struct eth_priv *) poa->priv, dst->eth.dst.mac, + POA_MGMT_EID, buf, len, true, &abstime); +} + +const struct poa_ops eth_poa_ops = { + .poa_attach = eth_attach, + .poa_detach = eth_detach, + .poa_start = eth_start, + .poa_stop = eth_stop, + .poa_send = eth_send, + .poa_send_mgmt = eth_send_mgmt, + .poa_query = eth_query, + .poa_mtu = eth_mtu, + .poa_qlen = eth_qlen, +#ifdef HAVE_RAW_SOCKETS + .poa_qpkts = eth_qpkts, +#endif +#ifndef HAVE_NETMAP + .poa_rib = eth_rib, +#endif + .poa_spec = eth_spec, + .poa_has_id = eth_has_id, + .poa_match = eth_match, + .poa_link_match = eth_link_match, + .mpl = POA_ETH_MPL +}; diff --git a/src/lib/poa/poa.c b/src/lib/poa/poa.c new file mode 100644 index 00000000..3ad17c4f --- /dev/null +++ b/src/lib/poa/poa.c @@ -0,0 +1,2515 @@ +/* + * Ouroboros - Copyright (C) 2016 - 2026 + * + * Points of attachment (PoA) - transport independent core + * + * Included by dev.c; uses dev.c statics (proc, flow_init, ...). + * + * Dimitri Staessens <dimitri@ouroboros.rocks> + * Sander Vrijders <sander@ouroboros.rocks> + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public License + * version 2.1 as published by the Free Software Foundation. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., http://www.fsf.org/about/contact/. + */ + +#define POA_MAX_EIDS PROC_MAX_FLOWS +#define POA_ALLOC_TIMEO 10000 /* ms, overall FLOW_REQ deadline */ +#define POA_RETRY_TIMEO 300 /* ms, FLOW_REQ retransmit period */ +/* Must fit a certificate chain: post-quantum ones are large. */ +#define POA_MGMT_BUFSZ POA_MGMT_FRAME_SIZE +#define POA_MGMT_QMAX 64 /* queued management frames per PoA */ +#define POA_PEND_TIMEO 10 /* s, reap a request that never completes */ +#define POA_SWEEP_TIMEO 1000 /* ms, sweep interval */ +#define POA_SWEEP_MAX 16 /* requests reaped per sweep */ +#define POA_DEFER_MAX 64 /* replies waiting for their flow id */ + +/* EWMA over 8 samples. */ +#define POA_AVG_SHIFT 3 +/* Queue cost is sampled every 64th packet: qlen is a syscall. */ +#define POA_COST_MASK 63 +/* Reuse a qlen read for this long; the mark moves on doublings. */ + +#define POA_RIB "poa" +/* Fits the RIB labels below with 20-digit counters. */ +#define POA_RIB_STRLEN 2048 + +enum poa_flow_state { + POA_FLOW_NULL = 0, + POA_FLOW_PENDING, + POA_FLOW_ALLOCATED, + POA_FLOW_DEAD +}; + +enum poa_state { + POA_NULL = 0, + POA_INIT, /* poa_init(); this process may attach */ + POA_RUNNING, /* the threads are up */ + POA_OPERATIONAL /* bootstrapped or enrolled in a layer */ +}; + +struct poa_flow { + struct poa * poa; + + int flow_id; + uint32_t eid; + uint32_t r_eid; + struct poa_addr r_addr; + + enum poa_flow_state state; + + struct ssm_rbuff * rx_rb; + + struct list_head pend; /* on poa->pending while unanswered */ + struct timespec t0; + + /* Our answer, kept to re-send when a request is repeated. */ + bool answered; + bool answer_sent; + + /* Handed to a caller that will attach it; not the sweeper's. */ + bool claimed; + int answer; + buffer_t answer_data; + + /* Handshake rendezvous with the reader thread. */ + pthread_mutex_t mtx; + pthread_cond_t cond; + bool replied; + bool pending; + int response; + buffer_t resp_data; +}; + +/* + * An answer can be ready before the flow it answers has an id, and + * the caller must not be kept waiting for one: it answers to the IRMd, + * which gives up long before we would. + */ +struct poa_deferred { + struct list_head next; + struct timespec t0; + int flow_id; + int response; + buffer_t data; +}; + +struct poa_mgmt_frame { + struct list_head next; + struct poa * poa; + struct poa_addr src; + size_t len; + uint8_t buf[POA_MGMT_BUFSZ]; +}; + +/* + * A detach is performed by the management thread, so that it cannot + * run alongside the accept side. The caller waits for the result. + */ +struct poa_detach_req { + struct list_head next; + struct poa_spec spec; + int result; + bool done; +}; + +static struct { + struct list_head list; + + struct poa_flow * id_to_pf[SYS_MAX_FLOWS]; + + struct llist deferred; + + pthread_mutex_t mtx; /* guards id_to_pf */ + pthread_cond_t cond; + + /* One management thread and one link monitor for all PoAs. */ + struct llist mgmt_frames; + struct list_head detach; + pthread_mutex_t mgmt_mtx; + pthread_cond_t mgmt_cond; /* work for the handler */ + pthread_cond_t done_cond; /* a detach has completed */ + pthread_t mgmt_handler; + bool mgmt_stop; + + pthread_t monitor; + int mon_fd; + + enum poa_state state; + + /* Hashes of the names this process answers queries for. */ + uint8_t name_hash[POA_QUERY_HLEN]; + uint8_t layer_hash[POA_QUERY_HLEN]; + + pthread_rwlock_t lock; /* the PoAs and their flows */ +} poas; + +static int mgmt_send(struct poa * poa, + const struct poa_addr * dst, + uint8_t code, + uint32_t s_eid, + uint32_t d_eid, + qosspec_t qs, + int response, + const buffer_t * data) +{ + uint8_t buf[POA_MGMT_BUFSZ]; + struct poa_mgmt_msg * msg = (struct poa_mgmt_msg *) buf; + size_t len; + + len = sizeof(*msg); + if (data != NULL && data->len > 0) { + if (len + data->len > sizeof(buf)) + return -EMSGSIZE; + memcpy(buf + len, data->data, data->len); + + len += data->len; + } + + poa_mgmt_msg_ser(msg, code, s_eid, d_eid, qs, response, + data != NULL ? data->len : 0); + + POA_STAT_BUMP(poa, mgmt_tx); + + return poa->ops->poa_send_mgmt(poa, dst, buf, len); +} + +/* Caller holds poas.lock for writing. */ +static struct poa_flow * pf_create(struct poa * poa, + const struct poa_addr * r_addr) +{ + struct poa_flow * pf; + pthread_condattr_t cattr; + int eid; + + pf = malloc(sizeof(*pf)); + if (pf == NULL) + goto fail_malloc; + + memset(pf, 0, sizeof(*pf)); + + if (pthread_mutex_init(&pf->mtx, NULL) != 0) + goto fail_mtx; + + if (pthread_condattr_init(&cattr) != 0) + goto fail_cond; +#ifndef __APPLE__ + pthread_condattr_setclock(&cattr, PTHREAD_COND_CLOCK); +#endif + if (pthread_cond_init(&pf->cond, &cattr) != 0) { + pthread_condattr_destroy(&cattr); + goto fail_cond; + } + + pthread_condattr_destroy(&cattr); + + eid = bmp_allocate(poa->eids); + if (!bmp_is_id_valid(poa->eids, eid)) + goto fail_eid; + + pf->poa = poa; + pf->eid = (uint32_t) eid; + pf->flow_id = -1; + pf->state = POA_FLOW_PENDING; + pf->r_addr = *r_addr; + + clock_gettime(PTHREAD_COND_CLOCK, &pf->t0); + + rcu_wrlock(&poa->guard); + + rcu_publish(pf); + rcu_assign(poa->eid_to_pf[eid], pf); + + rcu_wrunlock(&poa->guard); + + list_add_tail(&pf->pend, &poa->flows); + + POA_STAT_BUMP(poa, n_flows); + + return pf; + + fail_eid: + pthread_cond_destroy(&pf->cond); + fail_cond: + pthread_mutex_destroy(&pf->mtx); + fail_mtx: + free(pf); + fail_malloc: + return NULL; +} + +/* Caller holds poas.mtx. */ +static void deferred_purge(int flow_id) +{ + struct list_head * p; + struct list_head * h; + + list_for_each_safe(p, h, &poas.deferred.list) { + struct poa_deferred * d; + + d = list_entry(p, struct poa_deferred, next); + if (d->flow_id != flow_id) + continue; + + llist_del(&d->next, &poas.deferred); + freebuf(d->data); + free(d); + } +} + +/* The id may have moved to a newer flow; only its owner clears it. */ +static void pf_destroy(struct poa_flow * pf) +{ + struct poa * poa = pf->poa; + + pthread_rwlock_wrlock(&poas.lock); + + rcu_wrlock(&poa->guard); + + rcu_assign(poa->eid_to_pf[pf->eid], NULL); + rcu_assign(pf->rx_rb, NULL); + + rcu_wrunlock(&poa->guard); + + bmp_release(poa->eids, pf->eid); + + if (!list_is_empty(&pf->pend)) { + list_del(&pf->pend); + POA_STAT_SUB(poa, n_flows, 1); + } + + pthread_rwlock_unlock(&poas.lock); + + pthread_mutex_lock(&poas.mtx); + + if (pf->flow_id >= 0 && poas.id_to_pf[pf->flow_id] == pf) { + poas.id_to_pf[pf->flow_id] = NULL; + + deferred_purge(pf->flow_id); + } + + pthread_mutex_unlock(&poas.mtx); + + rcu_reclaim(&poa->guard); /* a receive may still hold pf */ + + freebuf(pf->resp_data); + freebuf(pf->answer_data); + + pthread_cond_destroy(&pf->cond); + pthread_mutex_destroy(&pf->mtx); + + free(pf); +} + +/* + * Keeps the answer for a repeat, then sends it. An accept must wait + * until the flow can receive; a refusal needs no receiver. + */ +static int pf_answer(struct poa_flow * pf, + int response, + const buffer_t * data) +{ + int err; + + pthread_rwlock_wrlock(&poas.lock); + + freebuf(pf->answer_data); + + if (data != NULL && data->len > 0) { + pf->answer_data.data = malloc(data->len); + if (pf->answer_data.data != NULL) { + memcpy(pf->answer_data.data, data->data, data->len); + + pf->answer_data.len = data->len; + } + } + + pf->answer = response; + pf->answered = true; + if (response == 0 && pf->state != POA_FLOW_ALLOCATED) { + pthread_rwlock_unlock(&poas.lock); + return 0; + } + + pf->answer_sent = true; + + pthread_rwlock_unlock(&poas.lock); + + err = mgmt_send(pf->poa, &pf->r_addr, POA_FLOW_REPLY, pf->eid, + pf->r_eid, qos_raw, response, data); + + if (err == -ETIMEDOUT || err == -EAGAIN) + err = 0; /* stored; a repeat request resends it */ + + return err; +} + +/* Takes an answer left for a flow that had no id yet. */ +static struct poa_deferred * deferred_take(int flow_id) +{ + struct list_head * p; + struct list_head * h; + + list_for_each_safe(p, h, &poas.deferred.list) { + struct poa_deferred * d; + + d = list_entry(p, struct poa_deferred, next); + if (d->flow_id != flow_id) + continue; + + llist_del(&d->next, &poas.deferred); + + return d; + } + + return NULL; +} + +/* Publishes the flow_id so an answer can find this flow. */ +static void pf_set_flow_id(struct poa_flow * pf, + int flow_id) +{ + struct poa_deferred * d; + + pthread_mutex_lock(&poas.mtx); + + pf->flow_id = flow_id; + poas.id_to_pf[flow_id] = pf; + + d = deferred_take(flow_id); + + pthread_cond_broadcast(&poas.cond); + + pthread_mutex_unlock(&poas.mtx); + + if (d != NULL) { + pf_answer(pf, d->response, &d->data); + freebuf(d->data); + free(d); + } +} + +/* + * Between the request arriving and the accept returning, the flow has + * an id but no fd yet; flow_init claims the PoA here. + */ +static void pf_set_pending(struct poa_flow * pf) +{ + pthread_mutex_lock(&poas.mtx); + + pf->pending = true; + + pthread_mutex_unlock(&poas.mtx); +} + +static void pf_clr_pending(struct poa_flow * pf) +{ + pthread_mutex_lock(&poas.mtx); + + pf->pending = false; + + pthread_mutex_unlock(&poas.mtx); +} + +/* A process that attaches no PoA has nothing pending. */ +struct poa_flow * poa_flow_take_pending(int flow_id) +{ + struct poa_flow * pf; + + if (poas.state == POA_NULL) + return NULL; + + if (flow_id < 0 || flow_id >= SYS_MAX_FLOWS) + return NULL; + + pthread_mutex_lock(&poas.mtx); + + pf = poas.id_to_pf[flow_id]; + if (pf != NULL && pf->pending) + pf->pending = false; + else + pf = NULL; + + pthread_mutex_unlock(&poas.mtx); + + return pf; +} + +static struct poa_flow * pf_get(int flow_id) +{ + struct poa_flow * pf; + + if (flow_id < 0 || flow_id >= SYS_MAX_FLOWS) + return NULL; + + pthread_mutex_lock(&poas.mtx); + + pf = poas.id_to_pf[flow_id]; + + pthread_mutex_unlock(&poas.mtx); + + return pf; +} + +#ifdef PROC_FLOW_STATS + +/* Caller holds poas.lock. */ +static struct poa * poa_by_rib_name(const char * name) +{ + struct list_head * p; + + list_for_each(p, &poas.list) { + struct poa * poa = list_entry(p, struct poa, next); + + if (strcmp(poa->name, name) == 0) + return poa; + } + + return NULL; +} + +static int poa_rib_read(const char * path, + char * buf, + size_t len) +{ + struct poa * poa; + const char * entry; + size_t qlen; + size_t avg; + size_t cost; + int size; + int ret; + + entry = strstr(path, RIB_SEPARATOR) + 1; + + if (len < POA_RIB_STRLEN) + return -1; + + pthread_rwlock_rdlock(&poas.lock); + + poa = poa_by_rib_name(entry); + if (poa == NULL) + goto fail; + + qlen = poa->ops->poa_qlen(poa); + avg = poa->avg_len; + cost = poa->avg_len + poa->avg_ovh; + + size = snprintf(buf, len, + "Active flows: %zu\n" + "Packets received: %zu\n" + "Bytes received: %zu\n" + "Packets sent: %zu\n" + "Bytes sent: %zu\n" + "Management frames rcvd: %zu\n" + "Management frames sent: %zu\n" + "Bad EID packets: %zu\n" + "Delivery (N+1) failures: %zu\n" + "Buffer alloc failures: %zu\n" + "Packet read failures: %zu\n" + "Packet send failures: %zu\n" + "Name queries sent: %zu\n" + "Name queries received: %zu\n" + "Name replies sent: %zu\n" + "Name replies received: %zu\n" + "Queued (transport): %zu\n" + "Queued (packets): %zu\n" + "Mean packet size (bytes): %zu\n" + "Mean packet cost: %zu\n", + POA_STAT_LOAD(poa, n_flows), + POA_STAT_LOAD(poa, rx_pkts), + POA_STAT_LOAD(poa, rx_bytes), + POA_STAT_LOAD(poa, tx_pkts), + POA_STAT_LOAD(poa, tx_bytes), + POA_STAT_LOAD(poa, mgmt_rx), + POA_STAT_LOAD(poa, mgmt_tx), + POA_STAT_LOAD(poa, bad_eid), + POA_STAT_LOAD(poa, dlv_fail), + POA_STAT_LOAD(poa, buf_fail), + POA_STAT_LOAD(poa, rcv_fail), + POA_STAT_LOAD(poa, snd_fail), + POA_STAT_LOAD(poa, qry_tx), + POA_STAT_LOAD(poa, qry_rx), + POA_STAT_LOAD(poa, rep_tx), + POA_STAT_LOAD(poa, rep_rx), + qlen, cost > 0 ? qlen / cost : 0, avg, cost); + if (size < 0 || (size_t) size >= len) + goto fail; + + if (poa->ops->poa_rib != NULL) { + ret = poa->ops->poa_rib(poa, buf + size, len - size); + if (ret < 0) + goto fail; + + size += ret; + } + + pthread_rwlock_unlock(&poas.lock); + + return size; + + fail: + pthread_rwlock_unlock(&poas.lock); + + return -1; +} + +static int poa_rib_readdir(char *** buf) +{ + struct list_head * p; + size_t n = 0; + int idx = 0; + + pthread_rwlock_rdlock(&poas.lock); + + list_for_each(p, &poas.list) + ++n; + + if (n == 0) { + *buf = NULL; + goto no_poas; + } + + *buf = malloc(sizeof(**buf) * n); + if (*buf == NULL) + goto fail_entries; + + list_for_each(p, &poas.list) { + struct poa * poa = list_entry(p, struct poa, next); + + (*buf)[idx] = malloc(strlen(poa->name) + 1); + if ((*buf)[idx] == NULL) + goto fail_entry; + + strcpy((*buf)[idx++], poa->name); + } + no_poas: + pthread_rwlock_unlock(&poas.lock); + + return idx; + + fail_entry: + while (idx-- > 0) + free((*buf)[idx]); + + free(*buf); + fail_entries: + pthread_rwlock_unlock(&poas.lock); + + return -ENOMEM; +} + +static int poa_rib_getattr(const char * path, + struct rib_attr * attr) +{ + (void) path; + + attr->size = POA_RIB_STRLEN; + attr->mtime = 0; + + return 0; +} + +static struct rib_ops poa_r_ops = { + .read = poa_rib_read, + .readdir = poa_rib_readdir, + .getattr = poa_rib_getattr +}; + +#endif /* PROC_FLOW_STATS */ + +int poa_init(const char * name) +{ + pthread_condattr_t cattr; + + assert(name != NULL); + + memset(&poas, 0, sizeof(poas)); + + str_hash(HASH_SHA3_256, poas.name_hash, name); + + poas.mon_fd = -1; + if (pthread_mutex_init(&poas.mtx, NULL) != 0) + goto fail_mtx; + + if (pthread_cond_init(&poas.cond, NULL) != 0) + goto fail_cond; + + if (pthread_mutex_init(&poas.mgmt_mtx, NULL) != 0) + goto fail_mgmt_mtx; + + if (pthread_condattr_init(&cattr) != 0) + goto fail_cattr; +#ifndef __APPLE__ + pthread_condattr_setclock(&cattr, PTHREAD_COND_CLOCK); +#endif + if (pthread_cond_init(&poas.mgmt_cond, &cattr) != 0) { + pthread_condattr_destroy(&cattr); + goto fail_cattr; + } + + pthread_condattr_destroy(&cattr); + + if (pthread_cond_init(&poas.done_cond, NULL) != 0) + goto fail_done_cond; + + if (pthread_rwlock_init(&poas.lock, NULL) != 0) + goto fail_lock; + + list_head_init(&poas.list); + llist_init(&poas.deferred); + llist_init(&poas.mgmt_frames); + list_head_init(&poas.detach); + + poas.state = POA_INIT; + +#ifdef PROC_FLOW_STATS + if (rib_reg(POA_RIB, &poa_r_ops) < 0) + goto fail_rib; +#endif + return 0; + +#ifdef PROC_FLOW_STATS + fail_rib: + pthread_rwlock_destroy(&poas.lock); +#endif + + fail_lock: + pthread_cond_destroy(&poas.done_cond); + fail_done_cond: + pthread_cond_destroy(&poas.mgmt_cond); + fail_cattr: + pthread_mutex_destroy(&poas.mgmt_mtx); + fail_mgmt_mtx: + pthread_cond_destroy(&poas.cond); + fail_cond: + pthread_mutex_destroy(&poas.mtx); + fail_mtx: + return -1; +} + +int poa_set_layer(const char * layer) +{ + if (layer == NULL) + return -EINVAL; + + pthread_rwlock_wrlock(&poas.lock); + + str_hash(HASH_SHA3_256, poas.layer_hash, layer); + + poas.state = POA_OPERATIONAL; + + pthread_rwlock_unlock(&poas.lock); + + return 0; +} + +bool poa_has_name(const uint8_t * hash) +{ + bool match = false; + + pthread_rwlock_rdlock(&poas.lock); + + if (poas.state >= POA_INIT) + match = memcmp(hash, poas.name_hash, POA_QUERY_HLEN) == 0; + + if (!match && poas.state >= POA_OPERATIONAL) + match = memcmp(hash, poas.layer_hash, POA_QUERY_HLEN) == 0; + + pthread_rwlock_unlock(&poas.lock); + + return match; +} + +int poa_spb_reserve(struct ssm_pk_buff ** spb, + size_t len) +{ + return ssm_pool_alloc_b(proc.pool, len, NULL, spb, NULL) < 0 ? -1 : 0; +} + +void poa_spb_release(struct ssm_pk_buff * spb) +{ + ssm_pool_remove(proc.pool, ssm_pk_buff_get_off(spb)); +} + +void poa_rx_pkt(struct poa * poa, + uint32_t eid, + struct ssm_pk_buff * spb) +{ + struct poa_flow * pf; + struct ssm_rbuff * rx_rb; + size_t len; + + len = ssm_pk_buff_len(spb); /* the ring write takes it over */ + + if (eid >= poa->n_eids) { + POA_STAT_BUMP(poa, bad_eid); + poa_spb_release(spb); + return; + } + + rcu_rdlock(&poa->guard); + + pf = rcu_deref(poa->eid_to_pf[eid]); + + rcu_consume(pf); + + if (pf == NULL) + goto fail_eid; + + rx_rb = rcu_deref(pf->rx_rb); + + rcu_consume(rx_rb); + + if (rx_rb == NULL) + goto fail_eid; + + if (ssm_rbuff_write(rx_rb, ssm_pk_buff_get_off(spb)) < 0) { + POA_STAT_BUMP(poa, dlv_fail); + rcu_rdunlock(&poa->guard); + poa_spb_release(spb); + return; + } + + POA_STAT_BUMP(poa, rx_pkts); + POA_STAT_ADD(poa, rx_bytes, len); + + ssm_flow_set_notify(proc.fqset, pf->flow_id, FLOW_PKT); + + rcu_rdunlock(&poa->guard); + + return; + + fail_eid: + POA_STAT_BUMP(poa, bad_eid); + + rcu_rdunlock(&poa->guard); + + poa_spb_release(spb); +} + +static int poa_flow_req_arr(struct poa_flow * pf, + qosspec_t qs, + uint32_t mtu, + const buffer_t * data) +{ + struct flow_info flow; + uint8_t buf[SOCK_BUF_SIZE]; + buffer_t msg = {SOCK_BUF_SIZE, buf}; + buffer_t out = BUF_INIT; + int err; + + memset(&flow, 0, sizeof(flow)); + + flow.n_pid = getpid(); + flow.n_1_pid = getpid(); + flow.qs = qs; + flow.mpl = pf->poa->mpl; + flow.mtu = mtu; + if (ipcp_poa_flow_req_arr__irm_req_ser(&msg, &flow, data) < 0) + return -ENOMEM; + + pf_set_pending(pf); + + err = send_recv_msg(&msg); + if (err < 0) + goto fail; + + err = poa_flow__irm_result_des(&msg, &flow, &out); + if (err < 0) + goto fail; + + freebuf(out); + + if (flow.id < 0 || flow.id >= SYS_MAX_FLOWS) { + err = -EBADF; + goto fail; + } + + pf_set_flow_id(pf, flow.id); + + return 0; + fail: + pf_clr_pending(pf); + return err; +} + +static void handle_flow_req(struct poa * poa, + const struct poa_addr * src, + const struct poa_mgmt_msg * msg, + const uint8_t * data, + size_t data_len) +{ + struct list_head * p; + struct poa_flow * pf = NULL; + qosspec_t qs; + buffer_t buf; + buffer_t answer_data = BUF_INIT; + uint32_t r_eid; + uint32_t mtu; + uint32_t eid = 0; + int answer = 0; + bool found = false; + bool served = false; + + r_eid = ntoh32(msg->s_eid); + + poa_mgmt_msg_qos(msg, &qs); + + pthread_rwlock_wrlock(&poas.lock); + + list_for_each(p, &poa->flows) { + pf = list_entry(p, struct poa_flow, pend); + if (pf->r_eid != r_eid || poa_addr_cmp(&pf->r_addr, src) != 0) + continue; + + answer = pf->answer; + eid = pf->eid; + served = true; + + found = pf->answer_sent; + if (found && pf->answer_data.len > 0) { + answer_data.data = malloc(pf->answer_data.len); + if (answer_data.data != NULL) { + memcpy(answer_data.data, pf->answer_data.data, + pf->answer_data.len); + answer_data.len = pf->answer_data.len; + } + } + break; + } + + if (!served) { + pf = pf_create(poa, src); + if (pf != NULL) + pf->r_eid = r_eid; + } + + pthread_rwlock_unlock(&poas.lock); + + if (found) { + mgmt_send(poa, src, POA_FLOW_REPLY, eid, r_eid, qos_raw, answer, + answer_data.len > 0 ? &answer_data : NULL); + freebuf(answer_data); + } + + if (served || pf == NULL) + return; + + buf.len = data_len; + buf.data = (uint8_t *) data; + mtu = poa->ops->poa_mtu(poa, src); + if (poa_flow_req_arr(pf, qs, mtu, &buf) < 0) { + mgmt_send(poa, src, POA_FLOW_REPLY, pf->eid, r_eid, qos_raw, + -1, NULL); + pf_destroy(pf); + } +} + +static void handle_flow_reply(struct poa * poa, + const struct poa_addr * src, + const struct poa_mgmt_msg * msg, + const uint8_t * data, + size_t data_len) +{ + struct poa_flow * pf; + uint32_t eid; + + eid = ntoh32(msg->d_eid); + + pthread_rwlock_rdlock(&poas.lock); + + pf = eid < poa->n_eids ? poa->eid_to_pf[eid] : NULL; + if (pf == NULL || pf->state != POA_FLOW_PENDING) { + pthread_rwlock_unlock(&poas.lock); + return; + } + + if (poa_addr_cmp(&pf->r_addr, src) != 0) { + pthread_rwlock_unlock(&poas.lock); + return; + } + + pthread_mutex_lock(&pf->mtx); + + if (pf->replied) { + pthread_mutex_unlock(&pf->mtx); + pthread_rwlock_unlock(&poas.lock); + return; + } + + if (data_len > 0) { + pf->resp_data.data = malloc(data_len); + if (pf->resp_data.data != NULL) { + memcpy(pf->resp_data.data, data, data_len); + + pf->resp_data.len = data_len; + } + } + + pf->r_eid = ntoh32(msg->s_eid); + pf->response = ntoh32(msg->response); + pf->replied = true; + + pthread_cond_broadcast(&pf->cond); + + pthread_mutex_unlock(&pf->mtx); + + pthread_rwlock_unlock(&poas.lock); +} + +static void handle_flow_update(struct poa * poa, + const struct poa_addr * src, + const struct poa_mgmt_msg * msg, + const uint8_t * data, + size_t data_len) +{ + struct poa_flow * pf; + buffer_t buf; + uint32_t eid; + + eid = ntoh32(msg->d_eid); + + pthread_rwlock_rdlock(&poas.lock); + + pf = eid < poa->n_eids ? poa->eid_to_pf[eid] : NULL; + if (pf == NULL || pf->state != POA_FLOW_ALLOCATED) { + pthread_rwlock_unlock(&poas.lock); + return; + } + + if (poa_addr_cmp(&pf->r_addr, src) != 0) { + pthread_rwlock_unlock(&poas.lock); + return; + } + + eid = (uint32_t) pf->flow_id; + + pthread_rwlock_unlock(&poas.lock); + + buf.len = data_len; + buf.data = (uint8_t *) data; + + ipcp_flow_update_arr((int) eid, &buf); +} + +static void mgmt_frame_handle(struct poa_mgmt_frame * frame) +{ + const struct poa_mgmt_msg * msg; + const uint8_t * data; + size_t data_len; + + msg = (const struct poa_mgmt_msg *) frame->buf; + if (frame->len < sizeof(*msg)) + return; + + data_len = ntoh16(msg->data_len); + if (data_len > frame->len - sizeof(*msg)) + return; + + data = frame->buf + sizeof(*msg); + + switch (msg->code) { + case POA_FLOW_REQ: + handle_flow_req(frame->poa, &frame->src, msg, data, data_len); + break; + case POA_FLOW_REPLY: + handle_flow_reply(frame->poa, &frame->src, msg, data, data_len); + break; + case POA_FLOW_UPDATE: + handle_flow_update(frame->poa, &frame->src, msg, data, + data_len); + break; + default: + break; + } +} + +static bool pf_steal(struct poa_flow * pf) +{ + bool stolen = false; + + pthread_mutex_lock(&poas.mtx); + + if (pf->pending) { + pf->pending = false; + poas.id_to_pf[pf->flow_id] = NULL; + + deferred_purge(pf->flow_id); + + stolen = true; + } + + pthread_mutex_unlock(&poas.mtx); + + return stolen; +} + +static void sweep_pending(void) +{ + struct poa_flow * dead[POA_SWEEP_MAX]; + struct list_head * p; + struct list_head * q; + struct timespec now; + size_t n = 0; + size_t i; + + clock_gettime(PTHREAD_COND_CLOCK, &now); + + pthread_rwlock_rdlock(&poas.lock); + + list_for_each(p, &poas.list) { + struct poa * poa = list_entry(p, struct poa, next); + + list_for_each(q, &poa->flows) { + struct poa_flow * pf; + + if (n == POA_SWEEP_MAX) + break; + + pf = list_entry(q, struct poa_flow, pend); + if (pf->state != POA_FLOW_PENDING || pf->claimed) + continue; + + if (now.tv_sec - pf->t0.tv_sec < POA_PEND_TIMEO) + continue; + + if (pf->flow_id >= 0 && !pf_steal(pf)) + continue; + + dead[n++] = pf; + } + + if (n == POA_SWEEP_MAX) + break; + } + + pthread_rwlock_unlock(&poas.lock); + + for (i = 0; i < n; ++i) + pf_destroy(dead[i]); + + pthread_mutex_lock(&poas.mtx); + + list_for_each_safe(p, q, &poas.deferred.list) { + struct poa_deferred * d; + + d = list_entry(p, struct poa_deferred, next); + if (now.tv_sec - d->t0.tv_sec < POA_PEND_TIMEO) + continue; + + llist_del(&d->next, &poas.deferred); + freebuf(d->data); + free(d); + } + + pthread_mutex_unlock(&poas.mtx); +} + +void poa_rx_mgmt(struct poa * poa, + const struct poa_addr * src, + const uint8_t * buf, + size_t len) +{ + struct poa_mgmt_frame * frame; + + if (len < sizeof(struct poa_mgmt_msg) || len > POA_MGMT_BUFSZ) + return; + + POA_STAT_BUMP(poa, mgmt_rx); + + frame = malloc(offsetof(struct poa_mgmt_frame, buf) + len); + if (frame == NULL) + return; + + frame->poa = poa; + frame->src = *src; + frame->len = len; + + memcpy(frame->buf, buf, len); + + pthread_mutex_lock(&poas.mgmt_mtx); + + if (poa->n_mgmt >= POA_MGMT_QMAX) { + pthread_mutex_unlock(&poas.mgmt_mtx); + free(frame); + return; + } + + ++poa->n_mgmt; + + llist_add_tail(&frame->next, &poas.mgmt_frames); + + pthread_cond_signal(&poas.mgmt_cond); + + pthread_mutex_unlock(&poas.mgmt_mtx); +} + +static void mgmt_frames_purge(const struct poa * poa) +{ + struct list_head * p; + struct list_head * h; + + pthread_mutex_lock(&poas.mgmt_mtx); + + list_for_each_safe(p, h, &poas.mgmt_frames.list) { + struct poa_mgmt_frame * frame; + + frame = list_entry(p, struct poa_mgmt_frame, next); + if (frame->poa != poa) + continue; + + --frame->poa->n_mgmt; + + llist_del(&frame->next, &poas.mgmt_frames); + + free(frame); + } + + pthread_mutex_unlock(&poas.mgmt_mtx); +} + +static bool poa_has_id(const struct poa * poa, + const struct poa_spec * spec) +{ + if (poa->type != spec->type) + return false; + + return poa->ops->poa_has_id(poa, spec); +} + +/* + * The PoA carrying dst, as the transport judges it. -EPERM if none + * matches, -EINVAL if several do. Caller holds poas.lock. + */ +static int poa_lookup(const struct poa_addr * dst, + struct poa ** out) +{ + struct list_head * p; + struct poa * found = NULL; + + list_for_each(p, &poas.list) { + struct poa * poa = list_entry(p, struct poa, next); + + if (poa->type != dst->type) + continue; + + if (!poa->ops->poa_match(poa, dst)) + continue; + + if (found != NULL) /* nothing given, two candidates */ + return -EINVAL; + + found = poa; + } + + if (found == NULL) + return -EPERM; + + *out = found; + + return 0; +} + +/* Sends are deadlined, bounding the lock hold on a full queue. */ +int poa_bcast_mgmt(const struct poa_addr * dst, + const uint8_t * buf, + size_t len) +{ + struct list_head * p; + int n = 0; + + pthread_rwlock_rdlock(&poas.lock); + + list_for_each(p, &poas.list) { + struct poa * poa = list_entry(p, struct poa, next); + + if (poa->type != dst->type) + continue; + + if (!poa->ops->poa_match(poa, dst)) + continue; + + if (poa->ops->poa_send_mgmt(poa, dst, buf, len) < 0) + continue; + + /* All management broadcasts are name queries. */ + POA_STAT_BUMP(poa, qry_tx); + ++n; + } + + pthread_rwlock_unlock(&poas.lock); + + return n; +} + +static bool deadline_is_malformed(const struct timespec * timeo) +{ + if (timeo == NULL) + return false; + + if (timeo->tv_sec < 0 || timeo->tv_nsec < 0) + return true; + + return timeo->tv_nsec >= BILLION; +} + +/* + * Complete addr for dst on any backend that can query. The ops are + * collected under poas.lock but called outside it: a query blocks up + * to its deadline and takes the lock again to broadcast. The tables + * are static, so nothing dangles; a struct poa cannot be carried + * across the unlock. The deadline applies per backend. + */ +int poa_query(const char * dst, + const struct timespec * timeo, + struct poa_addr * addr) +{ + const struct poa_ops * cand[POA_MAX_POAS]; + enum poa_type type[POA_MAX_POAS]; + struct list_head * p; + size_t n = 0; + size_t i; + int err = -ENOTSUP; + + if (dst == NULL || addr == NULL) + return -EINVAL; + + if (deadline_is_malformed(timeo)) + return -EINVAL; + + pthread_rwlock_rdlock(&poas.lock); + + list_for_each(p, &poas.list) { + struct poa * poa = list_entry(p, struct poa, next); + + if (poa->ops->poa_query == NULL) + continue; + + for (i = 0; i < n && cand[i] != poa->ops; i++) + ; + if (i < n) + continue; + + /* One type per backend: eth. Revisit if that changes. */ + cand[n] = poa->ops; + type[n++] = poa->type; + } + + pthread_rwlock_unlock(&poas.lock); + + for (i = 0; i < n; i++) { + memset(addr, 0, sizeof(*addr)); + + addr->type = type[i]; + + err = cand[i]->poa_query(dst, timeo, addr); + if (err == 0) + return 0; + } + + return err; +} + +static int poa_check(const struct poa_addr * dst) +{ + struct poa * poa; + int err; + + pthread_rwlock_rdlock(&poas.lock); + + err = poa_lookup(dst, &poa); + + pthread_rwlock_unlock(&poas.lock); + + return err; +} + +static int poa_alloc(const struct poa_addr * dst, + qosspec_t qs, + const buffer_t * req, + buffer_t * resp, + struct poa_flow ** pf_out, + uint32_t * mtu, + const struct timespec * timeo) +{ + struct timespec dflt = TIMESPEC_INIT_MS(POA_ALLOC_TIMEO); + struct timespec rintv = TIMESPEC_INIT_MS(POA_RETRY_TIMEO); + struct poa_flow * pf; + struct poa * poa; + struct timespec abstime; + struct timespec now; + struct timespec retry; + int err; + + pthread_rwlock_wrlock(&poas.lock); + + err = poa_lookup(dst, &poa); + if (err < 0) { + pthread_rwlock_unlock(&poas.lock); + return err; + } + + pf = pf_create(poa, dst); + if (pf == NULL) { + pthread_rwlock_unlock(&poas.lock); + return -ENOMEM; + } + + pf->claimed = true; + + pthread_rwlock_unlock(&poas.lock); + + clock_gettime(PTHREAD_COND_CLOCK, &abstime); + ts_add(&abstime, timeo != NULL ? timeo : &dflt, &abstime); + + pthread_mutex_lock(&pf->mtx); + + while (!pf->replied) { + pthread_mutex_unlock(&pf->mtx); + + err = mgmt_send(poa, dst, POA_FLOW_REQ, pf->eid, 0, qs, 0, req); + + pthread_mutex_lock(&pf->mtx); + + if (err < 0 && err != -ETIMEDOUT && err != -EAGAIN) { + err = -EIO; + goto fail; + } + + if (pf->replied) + break; + + clock_gettime(PTHREAD_COND_CLOCK, &now); + ts_add(&now, &rintv, &retry); + + if (ts_diff_ns(&retry, &abstime) > 0) + retry = abstime; + + pthread_cond_timedwait(&pf->cond, &pf->mtx, &retry); + + if (pf->replied) + break; + + clock_gettime(PTHREAD_COND_CLOCK, &now); + + if (ts_diff_ns(&now, &abstime) >= 0) { + err = -ETIMEDOUT; + goto fail; + } + } + + if (pf->response != 0) { + err = -ECONNREFUSED; + goto fail; + } + + *resp = pf->resp_data; + pf->resp_data.len = 0; + pf->resp_data.data = NULL; + + pthread_mutex_unlock(&pf->mtx); + + *mtu = poa->ops->poa_mtu(poa, dst); + *pf_out = pf; + + return 0; + + fail: + pthread_mutex_unlock(&pf->mtx); + pf_destroy(pf); + return err; +} + +static void poa_alloc_fail(struct poa_flow * pf) +{ + pf_destroy(pf); +} + +void poa_flow_attach(struct poa_flow * pf, + int flow_id, + struct ssm_rbuff * rx_rb) +{ + struct poa * poa = pf->poa; + + if (pf->flow_id != flow_id) + pf_set_flow_id(pf, flow_id); + + pthread_rwlock_wrlock(&poas.lock); + + pf->state = POA_FLOW_ALLOCATED; + + rcu_wrlock(&poa->guard); + + rcu_publish(rx_rb); + rcu_assign(pf->rx_rb, rx_rb); + + rcu_wrunlock(&poa->guard); + + pthread_rwlock_unlock(&poas.lock); +} + +void poa_flow_detach(struct poa_flow * pf) +{ + struct poa * poa = pf->poa; + + pthread_rwlock_wrlock(&poas.lock); + + pf->state = POA_FLOW_DEAD; + + rcu_wrlock(&poa->guard); + + rcu_assign(pf->rx_rb, NULL); + + rcu_wrunlock(&poa->guard); + + pthread_rwlock_unlock(&poas.lock); + + pf_destroy(pf); +} + +static size_t flows_updown(struct poa * poa, + bool up) +{ + struct list_head * p; + size_t n = 0; + + list_for_each(p, &poa->flows) { + struct poa_flow * pf; + struct flow * flow; + + pf = list_entry(p, struct poa_flow, pend); + if (pf->state != POA_FLOW_ALLOCATED || pf->flow_id < 0) + continue; + + flow = &proc.flows[proc.id_to_fd[pf->flow_id].fd]; + if (flow->info.id != pf->flow_id) + continue; + + if (((flow->oflags & FLOWFDOWN) != 0) != !up) + ++n; + + if (up) { + flow->oflags &= ~FLOWFDOWN; + + ssm_rbuff_clr_bits(flow->rx_rb, RB_FLOWDOWN); + } else { + flow->oflags |= FLOWFDOWN; + + ssm_rbuff_set_bits(flow->rx_rb, RB_FLOWDOWN); + } + + ssm_flow_set_notify(proc.fqset, pf->flow_id, + up ? FLOW_UP : FLOW_DOWN); + } + + return n; +} + +size_t poa_link_updown(int id, + bool up) +{ + struct list_head * p; + size_t n = 0; + + pthread_rwlock_wrlock(&proc.lock); + pthread_rwlock_rdlock(&poas.lock); + + list_for_each(p, &poas.list) { + struct poa * poa = list_entry(p, struct poa, next); + + if (poa->ops->poa_link_match == NULL) + continue; + + if (!poa->ops->poa_link_match(poa, id)) + continue; + + n += flows_updown(poa, up); + } + + pthread_rwlock_unlock(&poas.lock); + pthread_rwlock_unlock(&proc.lock); + + return n; +} + +static size_t poa_ewma(size_t avg, + size_t sz) +{ + if (avg == 0) + return sz; + + avg = avg + (sz >> POA_AVG_SHIFT) - (avg >> POA_AVG_SHIFT); + + return avg == 0 ? 1 : avg; +} + +static void poa_avg_len_update(struct poa * poa, + size_t sz) +{ + STORE_RELAXED(&poa->avg_len, poa_ewma(LOAD_RELAXED(&poa->avg_len), sz)); +} + +static size_t poa_qlen(struct poa * poa) +{ + struct timespec now; + uint64_t ns; + size_t qlen; + + clock_gettime(PTHREAD_COND_CLOCK, &now); + + ns = TS_TO_UINT64(now); + if (ns - LOAD_RELAXED(&poa->q_time) < POA_QLEN_GATE) + return LOAD_RELAXED(&poa->q_cache); + + qlen = poa->ops->poa_qlen(poa); + + STORE_RELAXED(&poa->q_cache, qlen); + STORE_RELAXED(&poa->q_time, ns); + + return qlen; +} + +static void poa_cost_sample(struct poa * poa, + size_t before, + size_t len) +{ + size_t after; + + after = poa->ops->poa_qlen(poa); + if (after <= before) + return; /* drained; nothing to learn */ + + after -= before; + if (after < len || after > (len << 2) + 1024) + return; + + STORE_RELAXED(&poa->avg_ovh, + poa_ewma(LOAD_RELAXED(&poa->avg_ovh), after - len)); +} + +int poa_flow_tx(struct poa_flow * pf, + struct ssm_pk_buff * spb, + bool block, + const struct timespec * abstime) +{ + struct poa * poa = pf->poa; + size_t len = ssm_pk_buff_len(spb); + size_t before = 0; + bool sample; + int ret; + + sample = (LOAD_RELAXED(&poa->n_tx) & POA_COST_MASK) == 0; + if (sample) + before = poa->ops->poa_qlen(poa); + + ret = poa->ops->poa_send(poa, &pf->r_addr, pf->r_eid, spb, block, + abstime); + if (ret < 0) { /* the caller releases the buffer */ + POA_STAT_BUMP(poa, snd_fail); + return ret; + } + + POA_STAT_BUMP(poa, tx_pkts); + POA_STAT_ADD(poa, tx_bytes, len); + + FETCH_ADD_RELAXED(&poa->n_tx, 1); + + poa_avg_len_update(poa, len); + + if (sample) + poa_cost_sample(poa, before, len); + + poa_spb_release(spb); + + return 0; +} + +size_t poa_flow_qlen(const struct poa_flow * pf) +{ + struct poa * poa = pf->poa; + uint64_t bytes; + size_t cost; + size_t pkts; + size_t byts; + + if (poa->ops->poa_qpkts != NULL && + poa->ops->poa_qpkts(poa, &pkts, &byts) == 0) + return byts; + + cost = LOAD_RELAXED(&poa->avg_len) + LOAD_RELAXED(&poa->avg_ovh); + if (LOAD_RELAXED(&poa->avg_ovh) == 0 || cost == 0) + return poa_qlen(poa); /* overstated beats false empty */ + + bytes = (uint64_t) poa_qlen(poa) * LOAD_RELAXED(&poa->avg_len); + + return (size_t) (bytes / cost); +} + +size_t poa_flow_qpkts(const struct poa_flow * pf) +{ + struct poa * poa = pf->poa; + size_t cost; + size_t pkts; + size_t byts; + + if (poa->ops->poa_qpkts != NULL && + poa->ops->poa_qpkts(poa, &pkts, &byts) == 0) + return pkts; + + cost = LOAD_RELAXED(&poa->avg_len) + LOAD_RELAXED(&poa->avg_ovh); + if (LOAD_RELAXED(&poa->avg_ovh) == 0 || cost == 0) + return 0; + + return poa_qlen(poa) / cost; +} + +size_t poa_flow_mean_len(const struct poa_flow * pf) +{ + return LOAD_RELAXED(&pf->poa->avg_len); +} + +int poa_flow_qid(const struct poa_flow * pf) +{ + return pf->poa->qid; +} + +void poa_flow_ready(struct poa_flow * pf) +{ + buffer_t data; + int answer; + + if (pf == NULL) + return; + + clrbuf(data); + + pthread_rwlock_wrlock(&poas.lock); + + if (!pf->answered || pf->answer_sent) { + pthread_rwlock_unlock(&poas.lock); + return; + } + + answer = pf->answer; + + if (pf->answer_data.len > 0) { + data.data = malloc(pf->answer_data.len); + if (data.data != NULL) { + memcpy(data.data, pf->answer_data.data, + pf->answer_data.len); + data.len = pf->answer_data.len; + } + } + + pf->answer_sent = true; + + pthread_rwlock_unlock(&poas.lock); + + mgmt_send(pf->poa, &pf->r_addr, POA_FLOW_REPLY, pf->eid, + pf->r_eid, qos_raw, answer, &data); + + freebuf(data); +} + +int poa_flow_alloc_resp(int flow_id, + int response, + const buffer_t * data) +{ + struct poa_deferred * d; + struct poa_flow * pf; + + if (flow_id < 0 || flow_id >= SYS_MAX_FLOWS) + return -EPERM; + + pthread_mutex_lock(&poas.mtx); + + pf = poas.id_to_pf[flow_id]; + if (pf != NULL) { + pthread_mutex_unlock(&poas.mtx); + return pf_answer(pf, response, data); + } + + if (poas.deferred.len >= POA_DEFER_MAX) { + pthread_mutex_unlock(&poas.mtx); + return -ENOMEM; + } + + d = malloc(sizeof(*d)); + if (d == NULL) { + pthread_mutex_unlock(&poas.mtx); + return -ENOMEM; + } + + memset(d, 0, sizeof(*d)); + + clock_gettime(PTHREAD_COND_CLOCK, &d->t0); + + d->flow_id = flow_id; + d->response = response; + + if (data != NULL && data->len > 0) { + d->data.data = malloc(data->len); + if (d->data.data == NULL) { + free(d); + pthread_mutex_unlock(&poas.mtx); + return -ENOMEM; + } + memcpy(d->data.data, data->data, data->len); + + d->data.len = data->len; + } + + llist_add_tail(&d->next, &poas.deferred); + + pthread_mutex_unlock(&poas.mtx); + + return 0; +} + +int poa_flow_update(int flow_id, + const buffer_t * data) +{ + struct poa_flow * pf; + + pf = pf_get(flow_id); + if (pf == NULL) + return -EPERM; + + return mgmt_send(pf->poa, &pf->r_addr, POA_FLOW_UPDATE, pf->eid, + pf->r_eid, qos_raw, 0, data); +} + +/* The PoA state is released when the flow itself is torn down. */ +int poa_flow_dealloc(int flow_id) +{ + (void) flow_id; + + return 0; +} + +/* PoA id 0 = management channel. */ +static struct poa * poa_create(enum poa_type type, + const struct poa_ops * ops, + size_t n_eids) +{ + struct poa * poa; + + poa = malloc(sizeof(*poa)); + if (poa == NULL) + goto fail_malloc; + + memset(poa, 0, sizeof(*poa)); + + poa->eid_to_pf = malloc(sizeof(*poa->eid_to_pf) * n_eids); + if (poa->eid_to_pf == NULL) + goto fail_map; + + memset(poa->eid_to_pf, 0, sizeof(*poa->eid_to_pf) * n_eids); + + poa->eids = bmp_create(n_eids - 1, 1); + if (poa->eids == NULL) + goto fail_bmp; + + if (rcu_guard_init(&poa->guard) != 0) + goto fail_guard; + + list_head_init(&poa->next); + list_head_init(&poa->flows); + + poa->type = type; + poa->ops = ops; + poa->mpl = ops->mpl; + poa->n_eids = n_eids; + poa->qid = -1; + + return poa; + + fail_guard: + bmp_destroy(poa->eids); + fail_bmp: + free(poa->eid_to_pf); + fail_map: + free(poa); + fail_malloc: + return NULL; +} + +static void poa_destroy(struct poa * poa) +{ + rcu_guard_fini(&poa->guard); + + bmp_destroy(poa->eids); + + free(poa->eid_to_pf); + free(poa); +} + +static void poa_teardown(struct poa * poa) +{ + if (poas.state >= POA_RUNNING) + poa->ops->poa_stop(poa); + + mgmt_frames_purge(poa); + + poa->ops->poa_detach(poa); + + poa_destroy(poa); +} + +static void poa_detach_all(void) +{ + pthread_rwlock_wrlock(&poas.lock); + + while (!list_is_empty(&poas.list)) { + struct poa * poa; + + poa = list_first_entry(&poas.list, struct poa, next); + + list_del(&poa->next); + + pthread_rwlock_unlock(&poas.lock); + + poa_teardown(poa); + + pthread_rwlock_wrlock(&poas.lock); + } + + pthread_rwlock_unlock(&poas.lock); +} + +static int poa_do_detach(const struct poa_detach_req * req) +{ + struct list_head * p; + struct poa * found = NULL; + + pthread_rwlock_wrlock(&poas.lock); + + list_for_each(p, &poas.list) { + struct poa * poa = list_entry(p, struct poa, next); + + if (!poa_has_id(poa, &req->spec)) + continue; + + found = poa; + break; + } + + if (found == NULL) { + pthread_rwlock_unlock(&poas.lock); + return -ENOENT; + } + + if (!list_is_empty(&found->flows)) { + pthread_rwlock_unlock(&poas.lock); + return -EBUSY; + } + + list_del(&found->next); + + pthread_rwlock_unlock(&poas.lock); + + poa_teardown(found); + + return 0; +} + +static int poa_del(const struct poa_spec * spec) +{ + struct poa_detach_req req; + int cs; + int ret; + + memset(&req, 0, sizeof(req)); + + req.spec = *spec; + + pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &cs); + + if (poas.state < POA_RUNNING) { /* set before workers run */ + ret = poa_do_detach(&req); + goto out; + } + + pthread_mutex_lock(&poas.mgmt_mtx); + + if (poas.mgmt_stop) { /* stopped: poa_fini reaps these */ + pthread_mutex_unlock(&poas.mgmt_mtx); + + ret = -EBUSY; + goto out; + } + + list_add_tail(&req.next, &poas.detach); + + pthread_cond_signal(&poas.mgmt_cond); + + while (!req.done) + pthread_cond_wait(&poas.done_cond, &poas.mgmt_mtx); + + pthread_mutex_unlock(&poas.mgmt_mtx); + + ret = req.result; + out: + pthread_setcancelstate(cs, NULL); + + return ret; +} + +static __inline__ bool mgmt_idle(void) +{ + if (poas.mgmt_stop) + return false; + + if (!llist_is_empty(&poas.mgmt_frames)) + return false; + + return list_is_empty(&poas.detach); +} + +static void detach_run(void) +{ + while (!list_is_empty(&poas.detach)) { + struct poa_detach_req * req; + + req = list_first_entry(&poas.detach, struct poa_detach_req, + next); + list_del(&req->next); + + pthread_mutex_unlock(&poas.mgmt_mtx); + + req->result = poa_do_detach(req); + + pthread_mutex_lock(&poas.mgmt_mtx); + + req->done = true; + + pthread_cond_broadcast(&poas.done_cond); + } +} + +static void * mgmt_handler(void * o) +{ + struct timespec intv = TIMESPEC_INIT_MS(POA_SWEEP_TIMEO); + + (void) o; + + while (true) { + struct poa_mgmt_frame * frame; + struct timespec abstime; + + pthread_mutex_lock(&poas.mgmt_mtx); + + detach_run(); + + while (mgmt_idle()) { + clock_gettime(PTHREAD_COND_CLOCK, &abstime); + ts_add(&abstime, &intv, &abstime); + + if (pthread_cond_timedwait(&poas.mgmt_cond, + &poas.mgmt_mtx, + &abstime) == ETIMEDOUT) { + pthread_mutex_unlock(&poas.mgmt_mtx); + sweep_pending(); + pthread_mutex_lock(&poas.mgmt_mtx); + } + } + + if (poas.mgmt_stop) { + detach_run(); /* nobody else serves these */ + + pthread_mutex_unlock(&poas.mgmt_mtx); + break; + } + + if (llist_is_empty(&poas.mgmt_frames)) { + pthread_mutex_unlock(&poas.mgmt_mtx); + continue; + } + + frame = llist_first_entry(&poas.mgmt_frames, + struct poa_mgmt_frame, next); + llist_del(&frame->next, &poas.mgmt_frames); + + --frame->poa->n_mgmt; + + pthread_mutex_unlock(&poas.mgmt_mtx); + + mgmt_frame_handle(frame); + + free(frame); + } + + return (void *) 0; +} + +#ifndef HAVE_ETH +/* Only the Ethernet transport reports link events. */ +int poa_monitor_open(void) +{ + return -1; +} + +void poa_monitor_read(int fd) +{ + (void) fd; +} +#endif + +static void * poa_monitor(void * o) +{ + (void) o; + + while (true) + poa_monitor_read(poas.mon_fd); + + return (void *) 0; +} + +static int monitor_start(void) +{ + poas.mon_fd = poa_monitor_open(); + if (poas.mon_fd < 0) + return 0; + + if (pthread_create(&poas.monitor, NULL, poa_monitor, NULL) == 0) + return 0; + + close(poas.mon_fd); + + poas.mon_fd = -1; + + return -1; +} + +static void monitor_stop(void) +{ + if (poas.mon_fd < 0) + return; + + pthread_cancel(poas.monitor); + pthread_join(poas.monitor, NULL); + + close(poas.mon_fd); + + poas.mon_fd = -1; +} + +int poa_start(void) +{ + struct list_head * p; + struct list_head * h; + + if (poas.state == POA_NULL) + return 0; + + if (pthread_create(&poas.mgmt_handler, NULL, mgmt_handler, NULL) != 0) + goto fail_mgmt; + + if (monitor_start() < 0) + goto fail_monitor; + + pthread_rwlock_rdlock(&poas.lock); + + list_for_each(p, &poas.list) { + struct poa * poa = list_entry(p, struct poa, next); + + if (poa->ops->poa_start(poa) < 0) + goto fail_reader; + } + + poas.state = POA_RUNNING; + + pthread_rwlock_unlock(&poas.lock); + + return 0; + + fail_reader: + list_for_each(h, &poas.list) { + struct poa * poa = list_entry(h, struct poa, next); + + if (h == p) + break; + + poa->ops->poa_stop(poa); + } + + pthread_rwlock_unlock(&poas.lock); + + monitor_stop(); + fail_monitor: + pthread_mutex_lock(&poas.mgmt_mtx); + + poas.mgmt_stop = true; + + pthread_cond_broadcast(&poas.mgmt_cond); + pthread_mutex_unlock(&poas.mgmt_mtx); + + pthread_join(poas.mgmt_handler, NULL); + fail_mgmt: + return -1; +} + +void poa_stop(void) +{ + struct list_head * p; + + if (poas.state < POA_RUNNING) + return; + + pthread_mutex_lock(&poas.mgmt_mtx); + + poas.mgmt_stop = true; + + pthread_cond_broadcast(&poas.mgmt_cond); + pthread_mutex_unlock(&poas.mgmt_mtx); + + pthread_join(poas.mgmt_handler, NULL); + + monitor_stop(); + + pthread_rwlock_rdlock(&poas.lock); + + list_for_each(p, &poas.list) { + struct poa * poa = list_entry(p, struct poa, next); + + poa->ops->poa_stop(poa); + } + + pthread_rwlock_unlock(&poas.lock); + + poas.state = POA_INIT; +} + +void poa_fini(void) +{ + if (poas.state == POA_NULL) + return; + + poa_stop(); + + poas.state = POA_NULL; + + poa_detach_all(); + +#ifdef PROC_FLOW_STATS + rib_unreg(POA_RIB); +#endif + + pthread_rwlock_destroy(&poas.lock); + pthread_cond_destroy(&poas.done_cond); + pthread_cond_destroy(&poas.mgmt_cond); + pthread_mutex_destroy(&poas.mgmt_mtx); + pthread_cond_destroy(&poas.cond); + pthread_mutex_destroy(&poas.mtx); +} + +/* + * Lowest queue id no attached PoA holds; detaching frees it by + * leaving the list. Caller holds poas.lock. + */ +static int poa_qid_alloc(void) +{ + struct list_head * p; + bool used[POA_MAX_POAS]; + int i; + + memset(used, 0, sizeof(used)); + + list_for_each(p, &poas.list) { + struct poa * poa = list_entry(p, struct poa, next); + + if (poa->qid >= 0 && poa->qid < POA_MAX_POAS) + used[poa->qid] = true; + } + + for (i = 0; i < POA_MAX_POAS; i++) + if (!used[i]) + return i; + + return -1; +} + +static int poa_add(const struct poa_spec * spec, + const struct poa_ops * ops, + size_t n_eids) +{ + struct list_head * p; + struct poa * poa; + int err; + + pthread_rwlock_rdlock(&poas.lock); + + list_for_each(p, &poas.list) { + poa = list_entry(p, struct poa, next); + if (poa_has_id(poa, spec)) { + pthread_rwlock_unlock(&poas.lock); + return -EPERM; + } + } + + pthread_rwlock_unlock(&poas.lock); + + poa = poa_create(spec->type, ops, n_eids); + if (poa == NULL) + return -ENOMEM; + + err = poa->ops->poa_attach(poa, spec); + if (err < 0) + goto fail_bind; + + err = poa_addr_name(&poa->local, poa->name, sizeof(poa->name)); + if (err < 0) + goto fail_start; + + err = -1; + + pthread_rwlock_wrlock(&poas.lock); + + poa->qid = poa_qid_alloc(); + if (poa->qid < 0) { + pthread_rwlock_unlock(&poas.lock); + goto fail_start; + } + + if (poas.state >= POA_RUNNING && poa->ops->poa_start(poa) < 0) { + pthread_rwlock_unlock(&poas.lock); + goto fail_start; + } + + list_add_tail(&poa->next, &poas.list); + + pthread_rwlock_unlock(&poas.lock); + + return 0; + + fail_start: + poa->ops->poa_detach(poa); + fail_bind: + poa_destroy(poa); + return err; +} + +/* The single place a type is bound to its transport. */ +int poa_attach(const struct poa_spec * poa) +{ + if (poa == NULL) + return -EINVAL; + + switch (poa->type) { + case POA_UDP4: + /* FALLTHRU */ + case POA_UDP6: + return poa_add(poa, &udp_poa_ops, POA_MAX_EIDS); + case POA_ETH: +#ifdef HAVE_ETH + return poa_add(poa, ð_poa_ops, POA_MAX_EIDS); +#else + return -ENOTSUP; +#endif + default: + return -ENOTSUP; + } +} + +int poa_detach(const struct poa_spec * poa) +{ + if (poa == NULL) + return -EINVAL; + + return poa_del(poa); +} + +ssize_t poa_list(struct poa_spec * specs, + size_t max) +{ + struct list_head * p; + size_t n = 0; + + if (specs == NULL) + return -EINVAL; + + pthread_rwlock_rdlock(&poas.lock); + + list_for_each(p, &poas.list) { + struct poa * poa = list_entry(p, struct poa, next); + + if (n++ >= max) + continue; + + memset(specs, 0, sizeof(*specs)); + + poa->ops->poa_spec(poa, specs); + + specs++; + } + + pthread_rwlock_unlock(&poas.lock); + + return (ssize_t) n; +} + +/* + * Complete peer for dst on the backend serving its type. The ops are + * borrowed under poas.lock and called outside it (see poa_query); if + * every PoA of the type detaches in between, the query's broadcast + * reaches nothing and reports -EPERM, as the lookup would. + */ +static int poa_peer_resolve(const char * dst, + struct poa_addr * peer) +{ + const struct poa_ops * ops = NULL; + struct list_head * p; + + pthread_rwlock_rdlock(&poas.lock); + + list_for_each(p, &poas.list) { + struct poa * poa = list_entry(p, struct poa, next); + + if (poa->type == peer->type) { + ops = poa->ops; + break; + } + } + + pthread_rwlock_unlock(&poas.lock); + + if (ops == NULL) /* nothing could carry the flow */ + return -EPERM; + + if (ops->poa_query == NULL) /* these addresses arrive complete */ + return 0; + + return ops->poa_query(dst, NULL, peer); +} + +/* + * Three steps: the IRMd creates the flow and prepares the key exchange, + * the PoA handshakes with the peer, the IRMd completes the + * exchange and hands us the key. + */ +int poa_flow_alloc(const char * dst, + const struct poa_addr * addr, + qosspec_t * qs, + const struct timespec * timeo) +{ + struct flow_info flow; + struct poa_flow * pf; + struct poa_addr peer; + struct crypt_sk crypt; + struct timespec t0; + struct timespec t1; + uint8_t key[SYMMKEYSZ]; + uint8_t buf[SOCK_BUF_SIZE]; + buffer_t msg = {SOCK_BUF_SIZE, buf}; + buffer_t req; + buffer_t resp; + uint32_t mtu = 0; + int err; + + if (addr == NULL) + return -EINVAL; + + if (qs != NULL && qs->service == SVC_STREAM && qs->loss != 0) + return -EINVAL; + + peer = *addr; + + err = poa_peer_resolve(dst, &peer); + if (err < 0) + return err; + + addr = &peer; + + err = poa_check(addr); + if (err < 0) + return err; + + memset(&flow, 0, sizeof(flow)); + + flow.n_pid = getpid(); + flow.n_1_pid = getpid(); + flow.qs = qs == NULL ? qos_raw : *qs; + if (poa_flow_alloc__irm_req_ser(&msg, &flow, dst) < 0) + return -ENOMEM; + + err = send_recv_msg(&msg); + if (err < 0) + return err; + + clrbuf(req); + clrbuf(resp); + + err = poa_flow__irm_result_des(&msg, &flow, &req); + if (err < 0) + return err; + + clock_gettime(PTHREAD_COND_CLOCK, &t0); + + err = poa_alloc(addr, flow.qs, &req, &resp, &pf, &mtu, timeo); + + freebuf(req); + + if (err < 0) + goto fail_alloc; + + clock_gettime(PTHREAD_COND_CLOCK, &t1); + + flow.mtu = mtu; + flow.mpl = pf->poa->mpl; + msg.len = SOCK_BUF_SIZE; + msg.data = buf; + if (poa_flow_alloc_r__irm_req_ser(&msg, &flow, &resp, 0) < 0) { + err = -ENOMEM; + goto fail_resp; + } + + freebuf(resp); + + err = send_recv_msg(&msg); + if (err < 0) + goto fail_msg; + + crypt.key = key; + crypt.epoch = 0; + crypt.role = CRYPT_ROLE_INIT; + + err = flow__irm_result_des(&msg, &flow, &crypt); + if (err < 0) + goto fail_msg; + + err = flow_init(&flow, &crypt, ts_diff_ns(&t1, &t0), pf); + + crypt_secure_clear(key, SYMMKEYSZ); + + if (err < 0) + goto fail_msg; + + if (qs != NULL) + *qs = flow.qs; + + return err; + + fail_resp: + freebuf(resp); + fail_msg: + poa_alloc_fail(pf); + return err; + fail_alloc: + msg.len = SOCK_BUF_SIZE; + msg.data = buf; + if (poa_flow_alloc_r__irm_req_ser(&msg, &flow, NULL, err) == 0) + send_recv_msg(&msg); + + return err; +} + diff --git a/src/lib/poa/poa.h b/src/lib/poa/poa.h new file mode 100644 index 00000000..014986a3 --- /dev/null +++ b/src/lib/poa/poa.h @@ -0,0 +1,364 @@ +/* + * Ouroboros - Copyright (C) 2016 - 2026 + * + * Points of attachment (PoA) - internal API + * + * Dimitri Staessens <dimitri@ouroboros.rocks> + * Sander Vrijders <sander@ouroboros.rocks> + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public License + * version 2.1 as published by the Free Software Foundation. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., http://www.fsf.org/about/contact/. + */ + +#ifndef OUROBOROS_LIB_POA_POA_H +#define OUROBOROS_LIB_POA_POA_H + +#include <ouroboros/atomics.h> +#include <ouroboros/ipcp-dev.h> +#include <ouroboros/list.h> +#include <ouroboros/qos.h> +#include <ouroboros/rcu.h> +#include <ouroboros/ssm_flow_set.h> +#include <ouroboros/ssm_pool.h> +#include <ouroboros/ssm_rbuff.h> +#include <ouroboros/time.h> +#include <ouroboros/utils.h> + +#include <errno.h> +#include <limits.h> +#include <poll.h> +#include <pthread.h> +#include <stdbool.h> +#include <stdint.h> + +#define POA_MGMT_EID 0 /* reserved for the mgmt channel */ + +#define POA_FLOW_REQ 1 +#define POA_FLOW_REPLY 2 +#define POA_FLOW_UPDATE 3 +#define POA_NAME_QUERY 4 +#define POA_NAME_REPLY 5 + +#define POA_QUERY_HLEN 32 /* SHA3-256, the query hash algorithm */ + +/* Fits "udp6.<ip6>.<port>", the longest display/RIB entry name. */ +#define POA_NAME_STRLEN 63 + +/* Management message; every transport frames it the same way. */ +struct poa_mgmt_msg { + uint8_t code; + uint8_t resv[3]; + uint32_t s_eid; + uint32_t d_eid; + int32_t response; + uint64_t bandwidth; + uint32_t delay; + uint32_t loss; + uint32_t ber; + uint32_t max_gap; + uint32_t timeout; + uint8_t availability; + uint8_t service; + uint16_t data_len; +} __attribute__((packed)); + +struct poa; +struct poa_flow; + +#ifdef PROC_FLOW_STATS +#define POA_STAT_BUMP(poa, field) FETCH_ADD_RELAXED(&(poa)->stat.field, 1) +#define POA_STAT_ADD(poa, field, v) FETCH_ADD_RELAXED(&(poa)->stat.field, (v)) +#define POA_STAT_SUB(poa, field, v) FETCH_SUB_RELAXED(&(poa)->stat.field, (v)) +#define POA_STAT_LOAD(poa, field) LOAD_RELAXED(&(poa)->stat.field) +#else +#define POA_STAT_BUMP(poa, field) ((void) (poa)) +#define POA_STAT_ADD(poa, field, v) ((void) (poa), (void) (v)) +#define POA_STAT_SUB(poa, field, v) ((void) (poa), (void) (v)) +#define POA_STAT_LOAD(poa, field) ((void) (poa), (size_t) 0) +#endif + +struct poa_stat { + size_t n_flows; /* gauge; the RIB reads it without poas.lock */ + size_t rx_pkts; /* packets delivered to a flow */ + size_t rx_bytes; /* payload bytes delivered */ + size_t tx_pkts; /* packets handed to the transport */ + size_t tx_bytes; /* payload bytes handed to the transport */ + size_t mgmt_rx; /* frames queued for the mgmt handler */ + size_t mgmt_tx; /* management frames sent */ + size_t bad_eid; /* no flow on the EID a peer sent */ + size_t dlv_fail; /* the rx ring above would not take it */ + size_t buf_fail; /* no buffer to receive into */ + size_t rcv_fail; /* transport read failed; the reader exits */ + size_t snd_fail; /* transport send failed */ + size_t qry_tx; /* name queries broadcast */ + size_t qry_rx; /* name queries received */ + size_t rep_tx; /* name replies sent, query matched */ + size_t rep_rx; /* name replies received */ +}; + +/* Spacing between transmit-depth samples; a depth costs a syscall. */ +#define POA_QLEN_GATE (100 * 1000) /* ns */ + + +/* Transport operations; public poa_X() dispatches to ops->poa_X. */ +struct poa_ops { + /* Parse own arm of the spec; validate; fill local and priv. */ + int (* poa_attach)(struct poa * poa, + const struct poa_spec * spec); + + void (* poa_detach)(struct poa * poa); + + int (* poa_start)(struct poa * poa); + + void (* poa_stop)(struct poa * poa); + + /* Full queue: -EAGAIN unless block; then wait, to abstime if set. */ + int (* poa_send)(struct poa * poa, + const struct poa_addr * dst, + uint32_t eid, + struct ssm_pk_buff * spb, + bool block, + const struct timespec * abstime); + + int (* poa_send_mgmt)(struct poa * poa, + const struct poa_addr * dst, + const uint8_t * buf, + size_t len); + + int (* poa_query)(const char * dst, + const struct timespec * timeo, + struct poa_addr * addr); + + uint32_t (* poa_mtu)(struct poa * poa, + const struct poa_addr * dst); + + /* Bytes queued in the transmit path of the PoA. */ + size_t (* poa_qlen)(struct poa * poa); + + /* Depth from the queue itself; NULL infers it from qlen. */ + int (* poa_qpkts)(struct poa * poa, + size_t * pkts, + size_t * byts); + + int (* poa_rib)(struct poa * poa, + char * buf, + size_t len); + + /* Identity as a spec, e.g. for poa_list. */ + void (* poa_spec)(const struct poa * poa, + struct poa_spec * spec); + + /* Same identity as spec? Caller matched poa->type already. */ + bool (* poa_has_id)(const struct poa * poa, + const struct poa_spec * spec); + + /* Carries dst? Caller matched poa->type already. */ + bool (* poa_match)(const struct poa * poa, + const struct poa_addr * dst); + + /* + * Flows ride the link this id names; NULL: no link events. + * Ids are meaningful only to the backend whose monitor + * produced them; a single backend owns the monitor. + */ + bool (* poa_link_match)(const struct poa * poa, + int id); + + /* Maximum packet lifetime in the transport, seconds. */ + time_t mpl; +}; +struct poa { + struct list_head next; + + enum poa_type type; + const struct poa_ops * ops; + void * priv; + + struct poa_addr local; /* what peers dial us on */ + + /* Display/RIB entry name for local, e.g. "udp4.<ip>.<port>". */ + char name[POA_NAME_STRLEN + 1]; + + time_t mpl; + + /* Identifies the transmit queue the flows on this PoA share. */ + int qid; + + /* Mean sent packet size (bytes), EWMA over the send path. */ + size_t avg_len; + /* Cost of one packet in the queue, in the transport's terms. */ + size_t avg_ovh; + size_t n_tx; + + /* Last queue depth read, and when, in the transport's terms. */ + size_t q_cache; + uint64_t q_time; + + /* Queued management frames, capped; poas.mgmt_mtx guards. */ + size_t n_mgmt; + +#ifdef PROC_FLOW_STATS + struct poa_stat stat; +#endif + + struct bmp * eids; + struct poa_flow ** eid_to_pf; + size_t n_eids; + + struct list_head flows; /* live flows, for repeats */ + + /* Keeps a flow and its ring alive under the receive path. */ + struct rcu_guard guard; +}; + +/* poa/poa.c is part of the dev.c translation unit. */ +int poa_init(const char * name); + +int poa_start(void); + +void poa_stop(void); + +void poa_fini(void); + +/* Also answer name queries for the layer once enrolled. */ +int poa_set_layer(const char * layer); + +/* Hash of a name this process answers queries for? */ +bool poa_has_name(const uint8_t * hash); + +int poa_flow_tx(struct poa_flow * pf, + struct ssm_pk_buff * spb, + bool block, + const struct timespec * abstime); + +size_t poa_flow_qlen(const struct poa_flow * pf); + +size_t poa_flow_qpkts(const struct poa_flow * pf); + +int poa_flow_qid(const struct poa_flow * pf); + +size_t poa_flow_mean_len(const struct poa_flow * pf); + +void poa_flow_attach(struct poa_flow * pf, + int flow_id, + struct ssm_rbuff * rx_rb); + +void poa_flow_ready(struct poa_flow * pf); + +void poa_flow_detach(struct poa_flow * pf); + +struct poa_flow * poa_flow_take_pending(int flow_id); + +/* Addresses and management messages (poa/addr.c). */ +int poa_addr_cmp(const struct poa_addr * a, + const struct poa_addr * b); + +/* Display/RIB entry name, e.g. "udp4.<ip>.<port>". */ +int poa_addr_name(const struct poa_addr * a, + char * buf, + size_t len); + +void poa_mgmt_msg_ser(struct poa_mgmt_msg * msg, + uint8_t code, + uint32_t s_eid, + uint32_t d_eid, + qosspec_t qs, + int response, + size_t data_len); + +void poa_mgmt_msg_qos(const struct poa_mgmt_msg * msg, + qosspec_t * qs); + +/* Called by the transports. */ +void poa_rx_pkt(struct poa * poa, + uint32_t eid, + struct ssm_pk_buff * spb); + +void poa_rx_mgmt(struct poa * poa, + const struct poa_addr * src, + const uint8_t * buf, + size_t len); + +/* Reserve a buffer for a received packet, with transport headroom. */ +int poa_spb_reserve(struct ssm_pk_buff ** spb, + size_t len); + +void poa_spb_release(struct ssm_pk_buff * spb); + +/* + * All flows on PoAs whose poa_link_match reports this link id go up + * or down with it. Returns the number of flows whose state changed. + */ +size_t poa_link_updown(int id, + bool up); + +/* + * Link monitor: one socket for the whole subsystem, opened by + * poa_start(). Returns -1 where the transport has no monitor. + */ +int poa_monitor_open(void); + +/* Reads one batch of link events; cancellation point. */ +void poa_monitor_read(int fd); + +/* Broadcast a mgmt frame on every PoA matching dst; # sent. */ +int poa_bcast_mgmt(const struct poa_addr * dst, + const uint8_t * buf, + size_t len); + +/* Transport op tables. */ +extern const struct poa_ops udp_poa_ops; +extern const struct poa_ops eth_poa_ops; + +/* + * Waits for a descriptor to take another packet, up to abstime. + * A NULL deadline waits indefinitely. Transports call this when + * their send reports the transmit queue full. + */ +static __inline__ int poa_wait_out(int fd, + const struct timespec * abstime) +{ + struct pollfd pfd; + struct timespec now; + long ms = -1; + bool clamped = false; + int ret; + + if (abstime != NULL) { + clock_gettime(PTHREAD_COND_CLOCK, &now); + + if (ts_diff_ns(abstime, &now) <= 0) + return -ETIMEDOUT; + + ms = ts_diff_ms(abstime, &now) + 1; /* sub-ms must wait */ + if (ms > INT_MAX) { /* poll takes an int */ + ms = INT_MAX; + clamped = true; + } + } + + pfd.fd = fd; + pfd.events = POLLOUT; + pfd.revents = 0; + + ret = poll(&pfd, 1, (int) ms); + if (ret < 0) + return errno == EINTR ? 0 : -EIO; + + if (ret == 0) + return clamped ? 0 : -ETIMEDOUT; /* clamped: retry */ + + return 0; +} + +#endif /* OUROBOROS_LIB_POA_POA_H */ diff --git a/src/lib/poa/udp.c b/src/lib/poa/udp.c new file mode 100644 index 00000000..6753347a --- /dev/null +++ b/src/lib/poa/udp.c @@ -0,0 +1,633 @@ +/* + * Ouroboros - Copyright (C) 2016 - 2026 + * + * Points of attachment (PoA) - UDP transport + * + * Dimitri Staessens <dimitri@ouroboros.rocks> + * Sander Vrijders <sander@ouroboros.rocks> + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public License + * version 2.1 as published by the Free Software Foundation. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., http://www.fsf.org/about/contact/. + */ + +#if defined(__APPLE__) +#define _BSD_SOURCE +#define _DARWIN_C_SOURCE +#elif defined(__FreeBSD__) +#define __BSD_VISIBLE 1 +#elif defined(__linux__) || defined(__CYGWIN__) +#ifndef _DEFAULT_SOURCE +#define _DEFAULT_SOURCE +#endif +#else +#ifndef _POSIX_C_SOURCE +#define _POSIX_C_SOURCE 200809L +#endif +#endif + +#include "config.h" + +#define OUROBOROS_PREFIX "poa-udp" + +#include <ouroboros/endian.h> +#include <ouroboros/errno.h> +#include <ouroboros/logs.h> +#include <ouroboros/time.h> +#include <ouroboros/utils.h> + +#include "poa.h" + +#ifdef __linux__ +#include <linux/sockios.h> +#endif +#include <arpa/inet.h> +#include <netinet/in.h> +#include <sys/ioctl.h> +#include <sys/socket.h> +#include <sys/uio.h> + +#include <errno.h> +#include <stdio.h> +#include <stdlib.h> +#include <string.h> +#include <unistd.h> + +#define UDP_HDR_LEN sizeof(uint32_t) /* PoA id */ +#define UDP_MAX_PAYLOAD (POA_UDP_RD_BUF - UDP_HDR_LEN) +/* The reader buffer must fit a full mgmt frame at any tuning. */ +#define UDP_MAX_PACKET MAX(POA_UDP_RD_BUF, POA_MGMT_FRAME_SIZE + UDP_HDR_LEN) +#define UDP_IP4_OVERH 28U /* IPv4 + UDP */ +#define UDP_IP6_OVERH 48U /* IPv6 + UDP */ +/* Wait for the link to come back before reading it again. */ +#define UDP_DOWN_TIMEO 100 /* ms */ + +union udp_saddr { + struct sockaddr sa; + struct sockaddr_in in; + struct sockaddr_in6 in6; +}; + +struct udp_priv { + int s_fd; + int af; + union udp_saddr s_saddr; + pthread_t reader; + bool running; +}; + +static socklen_t saddr_len(int af) +{ + if (af == AF_INET) + return sizeof(struct sockaddr_in); + + return sizeof(struct sockaddr_in6); +} + +static void addr_to_saddr(const struct poa_addr * addr, + union udp_saddr * saddr) +{ + memset(saddr, 0, sizeof(*saddr)); + + if (addr->type == POA_UDP4) { + saddr->in.sin_family = AF_INET; + saddr->in.sin_addr = addr->udp4.ip_addr; + saddr->in.sin_port = htons(addr->udp4.port); + } else { + saddr->in6.sin6_family = AF_INET6; + saddr->in6.sin6_addr = addr->udp6.ip_addr; + saddr->in6.sin6_port = htons(addr->udp6.port); + } +} + +static void saddr_to_addr(const union udp_saddr * saddr, + struct poa_addr * addr) +{ + memset(addr, 0, sizeof(*addr)); + + if (saddr->sa.sa_family == AF_INET) { + addr->type = POA_UDP4; + addr->udp4.ip_addr = saddr->in.sin_addr; + addr->udp4.port = ntohs(saddr->in.sin_port); + } else { + addr->type = POA_UDP6; + addr->udp6.ip_addr = saddr->in6.sin6_addr; + addr->udp6.port = ntohs(saddr->in6.sin6_port); + } +} + +/* A datagram longer than the buffer arrives truncated: drop it. */ +static void * udp_reader(void * o) +{ + struct poa * poa = (struct poa *) o; + struct udp_priv * priv = (struct udp_priv *) poa->priv; + struct timespec down = TIMESPEC_INIT_MS(UDP_DOWN_TIMEO); + uint8_t * buf; + + buf = malloc(UDP_MAX_PACKET); + if (buf == NULL) + return (void *) -1; + + pthread_cleanup_push(free, buf); + + while (true) { + struct ssm_pk_buff * spb; + union udp_saddr r_saddr; + struct poa_addr src; + struct msghdr mh; + struct iovec iov; + ssize_t n; + uint32_t eid; + size_t plen; + + iov.iov_base = buf; + iov.iov_len = UDP_MAX_PACKET; + + memset(&mh, 0, sizeof(mh)); + + mh.msg_name = &r_saddr; + mh.msg_namelen = sizeof(r_saddr); + mh.msg_iov = &iov; + mh.msg_iovlen = 1; + + n = recvmsg(priv->s_fd, &mh, 0); + if (n < 0) { + if (errno == EINTR) + continue; + + POA_STAT_BUMP(poa, rcv_fail); + + if (errno == ENETDOWN) { + nanosleep(&down, NULL); + continue; + } + + log_err("Reader stopped: %s.", strerror(errno)); + break; + } + + if ((mh.msg_flags & MSG_TRUNC) != 0) { + POA_STAT_BUMP(poa, buf_fail); + continue; + } + + if ((size_t) n < UDP_HDR_LEN) + continue; + + eid = ntoh32(*(uint32_t *) buf); + plen = (size_t) n - UDP_HDR_LEN; + + saddr_to_addr(&r_saddr, &src); + + if (eid == POA_MGMT_EID) { + poa_rx_mgmt(poa, &src, buf + UDP_HDR_LEN, plen); + continue; + } + + if (poa_spb_reserve(&spb, plen) < 0) { + POA_STAT_BUMP(poa, buf_fail); + continue; + } + + memcpy(ssm_pk_buff_head(spb), buf + UDP_HDR_LEN, plen); + + poa_rx_pkt(poa, eid, spb); + } + + pthread_cleanup_pop(true); + + return (void *) 0; +} + +/* Reads the bound address back: an ephemeral port is only known after. */ +static int udp_attach(struct poa * poa, + const struct poa_spec * spec) +{ + struct udp_priv * priv; + socklen_t len; + int af; + + af = spec->type == POA_UDP4 ? AF_INET : AF_INET6; + + priv = malloc(sizeof(*priv)); + if (priv == NULL) + return -ENOMEM; + + memset(priv, 0, sizeof(*priv)); + + priv->af = af; + priv->s_fd = socket(af, SOCK_DGRAM, IPPROTO_UDP); + if (priv->s_fd < 0) { + log_err("Failed to create socket: %s.", strerror(errno)); + goto fail_socket; + } + + if (af == AF_INET) { + priv->s_saddr.in.sin_family = AF_INET; + priv->s_saddr.in.sin_addr = spec->udp4.ip_addr; + priv->s_saddr.in.sin_port = htons(spec->udp4.port); + } else { + int on = 1; + + if (setsockopt(priv->s_fd, IPPROTO_IPV6, IPV6_V6ONLY, + &on, sizeof(on)) < 0) { + log_err("Failed to set IPV6_V6ONLY: %s.", + strerror(errno)); + goto fail_bind; + } + + priv->s_saddr.in6.sin6_family = AF_INET6; + priv->s_saddr.in6.sin6_addr = spec->udp6.ip_addr; + priv->s_saddr.in6.sin6_port = htons(spec->udp6.port); + } + + if (bind(priv->s_fd, &priv->s_saddr.sa, saddr_len(af)) < 0) { + log_err("Failed to bind: %s.", strerror(errno)); + goto fail_bind; + } + + poa->priv = priv; + len = saddr_len(af); + if (getsockname(priv->s_fd, &priv->s_saddr.sa, &len) < 0) + log_warn("Failed to read the bound address: %s.", + strerror(errno)); + + saddr_to_addr(&priv->s_saddr, &poa->local); + + return 0; + + fail_bind: + close(priv->s_fd); + fail_socket: + poa->priv = NULL; + + free(priv); + + return -EIO; +} + +static void udp_detach(struct poa * poa) +{ + struct udp_priv * priv = (struct udp_priv *) poa->priv; + + if (priv == NULL) + return; + + close(priv->s_fd); + + free(priv); + + poa->priv = NULL; +} + +/* One reader per socket, so a flow cannot be reordered on receive. */ +static int udp_start(struct poa * poa) +{ + struct udp_priv * priv = (struct udp_priv *) poa->priv; + + if (pthread_create(&priv->reader, NULL, udp_reader, poa) != 0) + return -1; + + priv->running = true; + + return 0; +} + +static void udp_stop(struct poa * poa) +{ + struct udp_priv * priv = (struct udp_priv *) poa->priv; + + if (!priv->running) + return; + + pthread_cancel(priv->reader); + pthread_join(priv->reader, NULL); + + priv->running = false; +} + +/* + * The PoA id is a separate iovec, so the buffer needs no headroom. + * MSG_DONTWAIT: the reader blocks on this socket. + */ +static int udp_sendv(struct udp_priv * priv, + const struct poa_addr * dst, + uint32_t eid, + const uint8_t * body, + size_t len, + bool block, + const struct timespec * abstime) +{ + union udp_saddr saddr; + struct msghdr msg; + struct iovec iov[2]; + uint32_t hdr; + int ret; + + if (len > UDP_MAX_PAYLOAD) + return -EMSGSIZE; + + addr_to_saddr(dst, &saddr); + + hdr = hton32(eid); + + iov[0].iov_base = &hdr; + iov[0].iov_len = sizeof(hdr); + iov[1].iov_base = (void *) body; + iov[1].iov_len = len; + + memset(&msg, 0, sizeof(msg)); + + msg.msg_name = &saddr; + msg.msg_namelen = saddr_len(priv->af); + msg.msg_iov = iov; + msg.msg_iovlen = len > 0 ? 2 : 1; + while (sendmsg(priv->s_fd, &msg, MSG_DONTWAIT) < 0) { + if (errno != EAGAIN && errno != EWOULDBLOCK) + return -EIO; + + if (!block) + return -EAGAIN; + + ret = poa_wait_out(priv->s_fd, abstime); + if (ret < 0) + return ret; + } + + return 0; +} + +static int udp_send(struct poa * poa, + const struct poa_addr * dst, + uint32_t eid, + struct ssm_pk_buff * spb, + bool block, + const struct timespec * abstime) +{ + return udp_sendv((struct udp_priv *) poa->priv, dst, eid, + ssm_pk_buff_head(spb), ssm_pk_buff_len(spb), + block, abstime); +} + +static int udp_send_mgmt(struct poa * poa, + const struct poa_addr * dst, + const uint8_t * buf, + size_t len) +{ + struct timespec timeo = TIMESPEC_INIT_MS(POA_MGMT_SND_TIMEO); + struct timespec abstime; + + clock_gettime(PTHREAD_COND_CLOCK, &abstime); + ts_add(&abstime, &timeo, &abstime); + + return udp_sendv((struct udp_priv *) poa->priv, dst, POA_MGMT_EID, + buf, len, true, &abstime); +} + +/* The PoA id header eats into the usable MTU. */ +static uint32_t udp_mtu(struct poa * poa, + const struct poa_addr * dst) +{ + struct udp_priv * priv = (struct udp_priv *) poa->priv; + uint32_t fallback; + uint32_t overh; +#if defined(__linux__) && (defined(IP_MTU) || defined(IPV6_MTU)) + union udp_saddr saddr; + socklen_t len; + int sock; + int mtu = 0; +#endif + if (priv->af == AF_INET) { + fallback = POA_UDP4_MTU; + overh = UDP_IP4_OVERH; + } else { + fallback = POA_UDP6_MTU; + overh = UDP_IP6_OVERH; + } + + fallback -= UDP_HDR_LEN; + if (fallback > UDP_MAX_PAYLOAD) + fallback = UDP_MAX_PAYLOAD; + +#if defined(__linux__) && (defined(IP_MTU) || defined(IPV6_MTU)) + + addr_to_saddr(dst, &saddr); + + sock = socket(priv->af, SOCK_DGRAM, IPPROTO_UDP); + if (sock < 0) + return fallback; + + if (connect(sock, &saddr.sa, saddr_len(priv->af)) < 0) + goto fallback; + + len = sizeof(mtu); + +#if defined(IP_MTU) + if (priv->af == AF_INET) { + if (getsockopt(sock, IPPROTO_IP, IP_MTU, &mtu, &len) < 0) + goto fallback; + } +#endif +#if defined(IPV6_MTU) + if (priv->af == AF_INET6) { + if (getsockopt(sock, IPPROTO_IPV6, IPV6_MTU, &mtu, &len) < 0) + goto fallback; + } +#endif + close(sock); + + if (mtu <= (int) (overh + UDP_HDR_LEN)) + return fallback; + + return MIN((uint32_t) mtu - overh - UDP_HDR_LEN, UDP_MAX_PAYLOAD); + + fallback: + close(sock); +#else + (void) dst; + (void) overh; +#endif + return fallback; +} + +/* All flows on the PoA share the socket, so this is aggregate. */ +static size_t udp_qlen(struct poa * poa) +{ +#if defined(__linux__) && defined(SIOCOUTQ) + struct udp_priv * priv = (struct udp_priv *) poa->priv; + int qlen; + + qlen = 0; + if (ioctl(priv->s_fd, SIOCOUTQ, &qlen) < 0) + return 0; + + return (size_t) qlen; +#else + (void) poa; + + return 0; +#endif +} + +/* The kernel keeps no per-socket drop count for UDP. */ +static int udp_rib(struct poa * poa, + char * buf, + size_t len) +{ + struct udp_priv * priv = (struct udp_priv *) poa->priv; + socklen_t optlen; + size_t sndbuf = 0; + size_t rcvbuf = 0; + int val; + int size; + + optlen = sizeof(val); + if (getsockopt(priv->s_fd, SOL_SOCKET, SO_SNDBUF, &val, &optlen) == 0) + sndbuf = (size_t) val; + + optlen = sizeof(val); + if (getsockopt(priv->s_fd, SOL_SOCKET, SO_RCVBUF, &val, &optlen) == 0) + rcvbuf = (size_t) val; + + size = snprintf(buf, len, + "Socket sndbuf (bytes): %zu\n" + "Socket rcvbuf (bytes): %zu\n", + sndbuf, rcvbuf); + if (size < 0 || (size_t) size >= len) + return -1; + + return size; +} + +/* + * Asks the kernel which address it would send from: connect() does the + * real route lookup and sends nothing, so this honours the default + * route, metrics and policy rules alike. + */ +static int udp_src_addr(const struct poa_addr * dst, + struct poa_addr * src) +{ + union udp_saddr saddr; + socklen_t len; + int af; + int fd; + + if (dst->type != POA_UDP4 && dst->type != POA_UDP6) + return -EINVAL; + + af = dst->type == POA_UDP4 ? AF_INET : AF_INET6; + + addr_to_saddr(dst, &saddr); + + fd = socket(af, SOCK_DGRAM, IPPROTO_UDP); + if (fd < 0) + return -EIO; + + if (connect(fd, &saddr.sa, saddr_len(af)) < 0) + goto fail; + + len = saddr_len(af); + if (getsockname(fd, &saddr.sa, &len) < 0) + goto fail; + + close(fd); + + saddr_to_addr(&saddr, src); + + return 0; + + fail: + close(fd); + + return -EIO; +} + +static void udp_spec(const struct poa * poa, + struct poa_spec * spec) +{ + spec->type = poa->type; + + if (poa->type == POA_UDP4) + spec->udp4 = poa->local.udp4; + else + spec->udp6 = poa->local.udp6; +} + +static bool udp_has_id(const struct poa * poa, + const struct poa_spec * spec) +{ + if (poa->type == POA_UDP4) { + if (poa->local.udp4.port != spec->udp4.port) + return false; + + return memcmp(&poa->local.udp4.ip_addr, + &spec->udp4.ip_addr, + sizeof(spec->udp4.ip_addr)) == 0; + } + + if (poa->local.udp6.port != spec->udp6.port) + return false; + + return memcmp(&poa->local.udp6.ip_addr, &spec->udp6.ip_addr, + sizeof(spec->udp6.ip_addr)) == 0; +} + +static bool udp_addr_is_any(const struct poa_addr * addr) +{ + static const struct in6_addr any6 = IN6ADDR_ANY_INIT; + + if (addr->type == POA_UDP4) + return addr->udp4.ip_addr.s_addr == htonl(INADDR_ANY); + + return memcmp(&addr->udp6.ip_addr, &any6, sizeof(any6)) == 0; +} + +/* + * Our end of the flow: the IP the kernel would send to dst from. A + * PoA bound to the wildcard is the catch-all and a failed probe + * matches any. Ports are not compared: the probe's is ephemeral. + */ +static bool udp_match(const struct poa * poa, + const struct poa_addr * dst) +{ + struct poa_addr src; + + if (udp_addr_is_any(&poa->local)) + return true; + + if (udp_src_addr(dst, &src) < 0) + return true; + + if (poa->type == POA_UDP4) + return memcmp(&poa->local.udp4.ip_addr, + &src.udp4.ip_addr, + sizeof(src.udp4.ip_addr)) == 0; + + return memcmp(&poa->local.udp6.ip_addr, &src.udp6.ip_addr, + sizeof(src.udp6.ip_addr)) == 0; +} + +const struct poa_ops udp_poa_ops = { + .poa_attach = udp_attach, + .poa_detach = udp_detach, + .poa_start = udp_start, + .poa_stop = udp_stop, + .poa_send = udp_send, + .poa_send_mgmt = udp_send_mgmt, + .poa_mtu = udp_mtu, + .poa_qlen = udp_qlen, + .poa_rib = udp_rib, + .poa_spec = udp_spec, + .poa_has_id = udp_has_id, + .poa_match = udp_match, + .mpl = POA_UDP_MPL +}; diff --git a/src/lib/protobuf.c b/src/lib/protobuf.c index 6beae000..aa247755 100644 --- a/src/lib/protobuf.c +++ b/src/lib/protobuf.c @@ -22,6 +22,7 @@ #define _DEFAULT_SOURCE +#include <ouroboros/errno.h> #include <ouroboros/protobuf.h> #include <ouroboros/crypt.h> #include <ouroboros/proc.h> @@ -215,6 +216,166 @@ struct layer_info layer_info_msg_to_s(const layer_info_msg_t * msg) return s; } + +static bool mac_is_zero(const uint8_t * mac) +{ + static const uint8_t zero[POA_MAC_SIZE] = { 0 }; + + return memcmp(mac, zero, POA_MAC_SIZE) == 0; +} + +poa_addr_msg_t * poa_addr_s_to_msg(const struct poa_addr * s) +{ + poa_addr_msg_t * msg; + + assert(s != NULL); + + msg = malloc(sizeof(*msg)); + if (msg == NULL) + goto fail_malloc; + + poa_addr_msg__init(msg); + + msg->type = s->type; + + if (s->hostname[0] != '\0') { + msg->hostname = strdup(s->hostname); + if (msg->hostname == NULL) + goto fail_msg; + } + + switch (s->type) { + case POA_UDP4: + msg->has_ip4 = true; + msg->ip4 = s->udp4.ip_addr.s_addr; + msg->has_port = true; + msg->port = s->udp4.port; + break; + case POA_UDP6: + msg->has_ip6 = true; + msg->ip6.len = sizeof(s->udp6.ip_addr); + msg->ip6.data = malloc(msg->ip6.len); + if (msg->ip6.data == NULL) + goto fail_msg; + memcpy(msg->ip6.data, &s->udp6.ip_addr, msg->ip6.len); + + msg->has_port = true; + msg->port = s->udp6.port; + break; + case POA_ETH: + msg->has_ethertype = true; + msg->ethertype = s->eth.dst.ethertype; + msg->has_mac = true; + msg->mac.len = POA_MAC_SIZE; + msg->mac.data = malloc(POA_MAC_SIZE); + if (msg->mac.data == NULL) + goto fail_msg; + memcpy(msg->mac.data, s->eth.dst.mac, POA_MAC_SIZE); + + if (!mac_is_zero(s->eth.src.mac)) { + msg->has_src_mac = true; + msg->src_mac.len = POA_MAC_SIZE; + msg->src_mac.data = malloc(POA_MAC_SIZE); + if (msg->src_mac.data == NULL) + goto fail_msg; + memcpy(msg->src_mac.data, s->eth.src.mac, POA_MAC_SIZE); + } + + if (s->eth.src.dev[0] != '\0') { + msg->dev = strdup(s->eth.src.dev); + if (msg->dev == NULL) + goto fail_msg; + } + break; + case POA_UDP: + msg->has_port = true; + msg->port = s->udp4.port; + break; + default: + goto fail_msg; + } + + return msg; + + fail_msg: + poa_addr_msg__free_unpacked(msg, NULL); + fail_malloc: + return NULL; +} + +struct poa_addr poa_addr_msg_to_s(const poa_addr_msg_t * msg) +{ + struct poa_addr s; + + assert(msg != NULL); + + memset(&s, 0, sizeof(s)); + + s.type = POA_INVALID; + + if (msg->hostname != NULL) { + if (strlen(msg->hostname) > POA_HOST_STRLEN) + return s; + strcpy(s.hostname, msg->hostname); + } + + switch (msg->type) { + case POA_UDP4: + if (!msg->has_ip4 || !msg->has_port) + return s; + + s.udp4.ip_addr.s_addr = msg->ip4; + s.udp4.port = msg->port; + break; + case POA_UDP6: + if (!msg->has_ip6 || !msg->has_port) + return s; + + if (msg->ip6.len != sizeof(s.udp6.ip_addr) + || msg->ip6.data == NULL) + return s; + + memcpy(&s.udp6.ip_addr, msg->ip6.data, msg->ip6.len); + + s.udp6.port = msg->port; + break; + case POA_ETH: + if (!msg->has_ethertype) + return s; + + s.eth.src.ethertype = msg->ethertype; + s.eth.dst.ethertype = msg->ethertype; + + if (msg->mac.len != POA_MAC_SIZE || msg->mac.data == NULL) + return s; + + memcpy(s.eth.dst.mac, msg->mac.data, POA_MAC_SIZE); + + if (msg->src_mac.len == POA_MAC_SIZE) + memcpy(s.eth.src.mac, msg->src_mac.data, POA_MAC_SIZE); + + if (msg->dev != NULL) { + if (strlen(msg->dev) > DEV_NAME_SIZE) + return s; + strcpy(s.eth.src.dev, msg->dev); + } + break; + case POA_UDP: + if (msg->hostname == NULL || msg->hostname[0] == '\0' + || !msg->has_port) + return s; + + s.udp4.port = msg->port; + break; + default: + return s; + } + + s.type = msg->type; + + return s; +} + ipcp_info_msg_t * ipcp_info_s_to_msg(const struct ipcp_info * s) { ipcp_info_msg_t * msg; @@ -389,22 +550,6 @@ struct dt_config dt_config_msg_to_s(const dt_config_msg_t * msg) return s; } -struct dir_dht_config dir_dht_config_msg_to_s(const dir_dht_config_msg_t * msg) -{ - struct dir_dht_config s; - - assert(msg != NULL); - - s.params.alpha = msg->alpha; - s.params.k = msg->k; - s.params.t_expire = msg->t_expire; - s.params.t_refresh = msg->t_refresh; - s.params.t_replicate = msg->t_replicate; - s.peer = msg->peer; - - return s; -} - dir_dht_config_msg_t * dir_dht_config_s_to_msg(const struct dir_dht_config * s) { dir_dht_config_msg_t * msg; @@ -427,22 +572,18 @@ dir_dht_config_msg_t * dir_dht_config_s_to_msg(const struct dir_dht_config * s) return msg; } -struct dir_config dir_config_msg_to_s(const dir_config_msg_t * msg) +struct dir_dht_config dir_dht_config_msg_to_s(const dir_dht_config_msg_t * msg) { - struct dir_config s; + struct dir_dht_config s; assert(msg != NULL); - switch (msg->pol) { - case DIR_DHT: - s.dht = dir_dht_config_msg_to_s(msg->dht); - break; - default: - /* No checks here */ - break; - } - - s.pol = msg->pol; + s.params.alpha = msg->alpha; + s.params.k = msg->k; + s.params.t_expire = msg->t_expire; + s.params.t_refresh = msg->t_refresh; + s.params.t_replicate = msg->t_replicate; + s.peer = msg->peer; return s; } @@ -479,6 +620,25 @@ dir_config_msg_t * dir_config_s_to_msg(const struct dir_config * s) return NULL; } +struct dir_config dir_config_msg_to_s(const dir_config_msg_t * msg) +{ + struct dir_config s; + + assert(msg != NULL); + + switch (msg->pol) { + case DIR_DHT: + s.dht = dir_dht_config_msg_to_s(msg->dht); + break; + default: + /* No checks here */ + break; + } + + s.pol = msg->pol; + + return s; +} uni_config_msg_t * uni_config_s_to_msg(const struct uni_config * s) { @@ -525,9 +685,9 @@ struct uni_config uni_config_msg_to_s(const uni_config_msg_t * msg) return s; } -udp4_config_msg_t * udp4_config_s_to_msg(const struct udp4_config * s) +udp4_poa_msg_t * udp4_poa_s_to_msg(const struct udp4_poa * s) { - udp4_config_msg_t * msg; + udp4_poa_msg_t * msg; assert(s != NULL); @@ -535,32 +695,30 @@ udp4_config_msg_t * udp4_config_s_to_msg(const struct udp4_config * s) if (msg == NULL) return NULL; - udp4_config_msg__init(msg); + udp4_poa_msg__init(msg); - msg->ip_addr = s->ip_addr.s_addr; - msg->dns_addr = s->dns_addr.s_addr; - msg->port = s->port; + msg->ip_addr = s->ip_addr.s_addr; + msg->port = s->port; return msg; } -struct udp4_config udp4_config_msg_to_s(const udp4_config_msg_t * msg) +struct udp4_poa udp4_poa_msg_to_s(const udp4_poa_msg_t * msg) { - struct udp4_config s; + struct udp4_poa s; assert(msg != NULL); - s.ip_addr.s_addr = msg->ip_addr; - s.dns_addr.s_addr = msg->dns_addr; - s.port = msg->port; + s.ip_addr.s_addr = msg->ip_addr; + s.port = msg->port; return s; } #define IN6_LEN (size_t) sizeof(struct in6_addr) -udp6_config_msg_t * udp6_config_s_to_msg(const struct udp6_config * s) +udp6_poa_msg_t * udp6_poa_s_to_msg(const struct udp6_poa * s) { - udp6_config_msg_t * msg; + udp6_poa_msg_t * msg; assert(s != NULL); @@ -568,7 +726,7 @@ udp6_config_msg_t * udp6_config_s_to_msg(const struct udp6_config * s) if (msg == NULL) goto fail_malloc; - udp6_config_msg__init(msg); + udp6_poa_msg__init(msg); msg->ip_addr.data = malloc(IN6_LEN); if (msg->ip_addr.data == NULL) @@ -577,42 +735,33 @@ udp6_config_msg_t * udp6_config_s_to_msg(const struct udp6_config * s) msg->ip_addr.len = IN6_LEN; memcpy(msg->ip_addr.data, &s->ip_addr.s6_addr, IN6_LEN); - msg->dns_addr.data = malloc(IN6_LEN); - if (msg->dns_addr.data == NULL) - goto fail_msg; - - msg->dns_addr.len = IN6_LEN; - memcpy(msg->dns_addr.data, &s->dns_addr.s6_addr, IN6_LEN); - msg->port = s->port; return msg; fail_msg: - udp6_config_msg__free_unpacked(msg, NULL); + udp6_poa_msg__free_unpacked(msg, NULL); fail_malloc: return NULL; } -struct udp6_config udp6_config_msg_to_s(const udp6_config_msg_t * msg) +struct udp6_poa udp6_poa_msg_to_s(const udp6_poa_msg_t * msg) { - struct udp6_config s; + struct udp6_poa s; assert(msg != NULL); assert(msg->ip_addr.len == IN6_LEN); - assert(msg->dns_addr.len == IN6_LEN); memcpy(&s.ip_addr.s6_addr, msg->ip_addr.data, IN6_LEN); - memcpy(&s.dns_addr.s6_addr, msg->dns_addr.data, IN6_LEN); s.port = msg->port; return s; } -eth_config_msg_t * eth_config_s_to_msg(const struct eth_config * s) +eth_poa_msg_t * eth_poa_s_to_msg(const struct eth_poa * s) { - eth_config_msg_t * msg; + eth_poa_msg_t * msg; assert(s != NULL); @@ -620,7 +769,7 @@ eth_config_msg_t * eth_config_s_to_msg(const struct eth_config * s) if (msg == NULL) goto fail_malloc; - eth_config_msg__init(msg); + eth_poa_msg__init(msg); msg->dev = strdup(s->dev); if (msg->dev == NULL) @@ -628,28 +777,115 @@ eth_config_msg_t * eth_config_s_to_msg(const struct eth_config * s) msg->ethertype = s->ethertype; + msg->has_mac = true; + msg->mac.len = POA_MAC_SIZE; + msg->mac.data = malloc(POA_MAC_SIZE); + if (msg->mac.data == NULL) + goto fail_msg; + + memcpy(msg->mac.data, s->mac, POA_MAC_SIZE); + return msg; fail_msg: - eth_config_msg__free_unpacked(msg, NULL); + eth_poa_msg__free_unpacked(msg, NULL); fail_malloc: return NULL; } -struct eth_config eth_config_msg_to_s(const eth_config_msg_t * msg) +struct eth_poa eth_poa_msg_to_s(const eth_poa_msg_t * msg) { - struct eth_config s; + struct eth_poa s; assert(msg != NULL); assert(strlen(msg->dev) <= DEV_NAME_SIZE); + memset(&s, 0, sizeof(s)); + strcpy(s.dev, msg->dev); + s.ethertype = msg->ethertype; + if (msg->mac.len == POA_MAC_SIZE) + memcpy(s.mac, msg->mac.data, POA_MAC_SIZE); + return s; } +poa_spec_msg_t * poa_spec_s_to_msg(const struct poa_spec * s) +{ + poa_spec_msg_t * msg; + + assert(s != NULL); + + msg = malloc(sizeof(*msg)); + if (msg == NULL) + goto fail_malloc; + + poa_spec_msg__init(msg); + + switch (s->type) { + case POA_UDP4: + msg->udp4 = udp4_poa_s_to_msg(&s->udp4); + if (msg->udp4 == NULL) + goto fail_msg; + break; + case POA_UDP6: + msg->udp6 = udp6_poa_s_to_msg(&s->udp6); + if (msg->udp6 == NULL) + goto fail_msg; + break; + case POA_ETH: + msg->eth = eth_poa_s_to_msg(&s->eth); + if (msg->eth == NULL) + goto fail_msg; + break; + default: + goto fail_msg; + } + + return msg; + + fail_msg: + poa_spec_msg__free_unpacked(msg, NULL); + fail_malloc: + return NULL; +} + +struct poa_spec poa_spec_msg_to_s(const poa_spec_msg_t * msg) +{ + struct poa_spec s; + + memset(&s, 0, sizeof(s)); + + s.type = POA_INVALID; + + if (msg == NULL) + return s; + + if (msg->udp4 != NULL) { + s.type = POA_UDP4; + s.udp4 = udp4_poa_msg_to_s(msg->udp4); + } else if (msg->udp6 != NULL) { + if (msg->udp6->ip_addr.len != IN6_LEN + || msg->udp6->ip_addr.data == NULL) + return s; + + s.type = POA_UDP6; + s.udp6 = udp6_poa_msg_to_s(msg->udp6); + } else if (msg->eth != NULL) { + if (msg->eth->dev == NULL + || strlen(msg->eth->dev) > DEV_NAME_SIZE) + return s; + + s.type = POA_ETH; + s.eth = eth_poa_msg_to_s(msg->eth); + } + + return s; +} + ipcp_config_msg_t * ipcp_config_s_to_msg(const struct ipcp_config * s) { ipcp_config_msg_t * msg; @@ -672,23 +908,6 @@ ipcp_config_msg_t * ipcp_config_s_to_msg(const struct ipcp_config * s) break; case IPCP_BROADCAST: break; - case IPCP_ETH_LLC: - /* FALLTHRU */ - case IPCP_ETH_DIX: - msg->eth = eth_config_s_to_msg(&s->eth); - if (msg->eth == NULL) - goto fail_msg; - break; - case IPCP_UDP4: - msg->udp4 = udp4_config_s_to_msg(&s->udp4); - if (msg->udp4 == NULL) - goto fail_msg; - break; - case IPCP_UDP6: - msg->udp6 = udp6_config_s_to_msg(&s->udp6); - if (msg->udp6 == NULL) - goto fail_msg; - break; default: /* No checks here */ break; @@ -714,6 +933,8 @@ struct ipcp_config ipcp_config_msg_to_s(const ipcp_config_msg_t * msg) assert(msg != NULL); + memset(&s, 0, sizeof(s)); + s.type = msg->ipcp_type; s.layer_info = layer_info_msg_to_s(msg->layer_info); @@ -724,17 +945,6 @@ struct ipcp_config ipcp_config_msg_to_s(const ipcp_config_msg_t * msg) case IPCP_UNICAST: s.unicast = uni_config_msg_to_s(msg->unicast); break; - case IPCP_ETH_LLC: - /* FALLTHRU */ - case IPCP_ETH_DIX: - s.eth = eth_config_msg_to_s(msg->eth); - break; - case IPCP_UDP4: - s.udp4 = udp4_config_msg_to_s(msg->udp4); - break; - case IPCP_UDP6: - s.udp6 = udp6_config_msg_to_s(msg->udp6); - break; case IPCP_BROADCAST: break; default: diff --git a/src/lib/serdes-irm.c b/src/lib/serdes-irm.c index 1d9b4dec..20b25a1b 100644 --- a/src/lib/serdes-irm.c +++ b/src/lib/serdes-irm.c @@ -441,6 +441,7 @@ int proc_exit__irm_req_ser(buffer_t * buf) return -ENOMEM; } +/* data is borrowed from the caller; detach before free. */ int ipcp_flow_req_arr__irm_req_ser(buffer_t * buf, const buffer_t * dst, const struct flow_info * flow, @@ -494,6 +495,138 @@ int ipcp_flow_req_arr__irm_req_ser(buffer_t * buf, return -ENOMEM; } +static int __ep_flow_ser(buffer_t * buf, + const struct flow_info * flow, + const buffer_t * data, + int response, + const char * dst, + int code) +{ + irm_msg_t * msg; + size_t len; + int err = -ENOMEM; + + msg = malloc(sizeof(*msg)); + if (msg == NULL) + goto fail_malloc; + + irm_msg__init(msg); + + msg->code = code; + msg->flow_info = flow_info_s_to_msg(flow); + if (msg->flow_info == NULL) + goto fail_msg; + + if (dst != NULL) { + msg->dst = strdup(dst); + if (msg->dst == NULL) + goto fail_msg; + } + + if (data != NULL) { + msg->has_pk = true; + msg->pk.len = data->len; + msg->pk.data = data->data; + } + + msg->has_response = true; + msg->response = response; + + 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); + + err = 0; + fail_msg: + msg->pk.len = 0; + msg->pk.data = NULL; + + irm_msg__free_unpacked(msg, NULL); + fail_malloc: + return err; +} + +int poa_flow_alloc__irm_req_ser(buffer_t * buf, + const struct flow_info * flow, + const char * dst) +{ + return __ep_flow_ser(buf, flow, NULL, 0, dst, + IRM_MSG_CODE__IRM_POA_FLOW_ALLOC); +} + +int poa_flow_alloc_r__irm_req_ser(buffer_t * buf, + const struct flow_info * flow, + const buffer_t * data, + int response) +{ + return __ep_flow_ser(buf, flow, data, response, NULL, + IRM_MSG_CODE__IRM_POA_FLOW_ALLOC_R); +} + +int ipcp_poa_flow_req_arr__irm_req_ser(buffer_t * buf, + const struct flow_info * flow, + const buffer_t * data) +{ + return __ep_flow_ser(buf, flow, data, 0, NULL, + IRM_MSG_CODE__IPCP_POA_FLOW_REQ_ARR); +} + +int poa_flow__irm_result_des(buffer_t * buf, + struct flow_info * flow, + buffer_t * data) +{ + irm_msg_t * msg; + int err; + + msg = irm_msg__unpack(NULL, buf->len, buf->data); + if (msg == NULL) { + err = -EIRMD; + goto fail_msg; + } + + if (!msg->has_result) { + err = -EIRMD; + goto fail; + } + + if (msg->result < 0) { + err = msg->result; + goto fail; + } + + if (msg->flow_info == NULL) { + err = -EBADF; + goto fail; + } + + *flow = flow_info_msg_to_s(msg->flow_info); + + clrbuf(*data); + + if (msg->has_pk && msg->pk.len > 0) { + data->data = malloc(msg->pk.len); + if (data->data == NULL) { + err = -ENOMEM; + goto fail; + } + memcpy(data->data, msg->pk.data, msg->pk.len); + + data->len = msg->pk.len; + } + + irm_msg__free_unpacked(msg, NULL); + + return 0; + fail: + irm_msg__free_unpacked(msg, NULL); + fail_msg: + return err; +} + int ipcp_flow_update_arr__irm_req_ser(buffer_t * buf, const struct flow_info * flow, const buffer_t * data) diff --git a/src/lib/tests/CMakeLists.txt b/src/lib/tests/CMakeLists.txt index 002d94af..1f2e9ba2 100644 --- a/src/lib/tests/CMakeLists.txt +++ b/src/lib/tests/CMakeLists.txt @@ -11,6 +11,7 @@ create_test_sourcelist(${PARENT_DIR}_tests test_suite.c bitmap_test.c btree_test.c crypt_test.c + poa_test.c hash_test.c kex_test.c kex_test_ml_kem.c @@ -25,6 +26,12 @@ create_test_sourcelist(${PARENT_DIR}_tests test_suite.c add_executable(${PARENT_DIR}_test ${${PARENT_DIR}_tests}) +if(HAVE_LIBURCU) + # poa_test.c pulls in poa.h, whose urcu guard needs C99. + set_source_files_properties(poa_test.c PROPERTIES + COMPILE_OPTIONS "-std=gnu99") +endif() + target_include_directories(${PARENT_DIR}_test PRIVATE ${CMAKE_SOURCE_DIR}/src/lib) diff --git a/src/lib/tests/poa_test.c b/src/lib/tests/poa_test.c new file mode 100644 index 00000000..99886769 --- /dev/null +++ b/src/lib/tests/poa_test.c @@ -0,0 +1,307 @@ +/* + * Ouroboros - Copyright (C) 2016 - 2026 + * + * Flow PoA tests + * + * Dimitri Staessens <dimitri@ouroboros.rocks> + * Sander Vrijders <sander@ouroboros.rocks> + * + * This program is free software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License version 2 as + * published by the Free Software Foundation. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program; if not, write to the Free Software + * Foundation, Inc., http://www.fsf.org/about/contact/. + */ + +#if defined(__linux__) || defined(__CYGWIN__) +#define _DEFAULT_SOURCE +#else +#define _POSIX_C_SOURCE 200809L +#endif + +#include <test/test.h> + +#include "poa/addr.c" +#ifdef HAVE_RAW_SOCKETS +#include "poa/eth.c" +#endif + +#include <arpa/inet.h> +#include <stdio.h> + +#ifdef HAVE_RAW_SOCKETS +static const uint8_t eth_our_mac[POA_MAC_SIZE] = + { 0x02, 0x00, 0x00, 0x00, 0x00, 0x01 }; +static const uint8_t eth_far_mac[POA_MAC_SIZE] = + { 0x02, 0x00, 0x00, 0x00, 0x00, 0x02 }; +static const uint8_t eth_bc_mac[POA_MAC_SIZE] = + { 0xff, 0xff, 0xff, 0xff, 0xff, 0xff }; +#endif + +/* PoA core stubs: the reader threads never run in this test. */ +void poa_rx_pkt(struct poa * poa, + uint32_t eid, + struct ssm_pk_buff * spb) +{ + (void) poa; + (void) eid; + (void) spb; +} + +void poa_rx_mgmt(struct poa * poa, + const struct poa_addr * src, + const uint8_t * buf, + size_t len) +{ + (void) poa; + (void) src; + (void) buf; + (void) len; +} + +int poa_spb_reserve(struct ssm_pk_buff ** spb, + size_t len) +{ + (void) spb; + (void) len; + + return -1; +} + +size_t poa_link_updown(int ifindex, + bool up) +{ + (void) ifindex; + (void) up; + + return 0; +} + +bool poa_has_name(const uint8_t * hash) +{ + (void) hash; + + return false; +} + +int poa_bcast_mgmt(const struct poa_addr * dst, + const uint8_t * buf, + size_t len) +{ + (void) dst; + (void) buf; + (void) len; + + return 0; +} + +static void udp4_addr(struct poa_addr * addr, + const char * ip, + uint16_t port) +{ + memset(addr, 0, sizeof(*addr)); + + addr->type = POA_UDP4; + addr->udp4.port = port; + + inet_pton(AF_INET, ip, &addr->udp4.ip_addr); +} + +static int test_poa_addr_cmp(void) +{ + struct poa_addr a; + struct poa_addr b; + + TEST_START(); + + udp4_addr(&a, "10.0.0.10", 3435); + udp4_addr(&b, "10.0.0.10", 3435); + + if (poa_addr_cmp(&a, &b) != 0) { + printf("Identical addresses did not match.\n"); + goto fail; + } + + udp4_addr(&b, "10.0.0.10", 3436); + + if (poa_addr_cmp(&a, &b) == 0) { + printf("Addresses with a different port matched.\n"); + goto fail; + } + + udp4_addr(&b, "10.0.0.11", 3435); + + if (poa_addr_cmp(&a, &b) == 0) { + printf("Addresses with a different ip matched.\n"); + goto fail; + } + + b.type = POA_UDP6; + if (poa_addr_cmp(&a, &b) == 0) { + printf("Addresses of a different type matched.\n"); + goto fail; + } + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + fail: + TEST_FAIL(); + return TEST_RC_FAIL; +} + +static int test_poa_mgmt_msg_qos(void) +{ + struct poa_mgmt_msg msg; + qosspec_t qs; + + TEST_START(); + + poa_mgmt_msg_ser(&msg, POA_FLOW_REQ, 1, 2, qos_stream, 0, 0); + + if (msg.code != POA_FLOW_REQ) { + printf("Wrong code in management message.\n"); + goto fail; + } + + if (ntoh32(msg.s_eid) != 1 || ntoh32(msg.d_eid) != 2) { + printf("Wrong PoA ids in management message.\n"); + goto fail; + } + + memset(&qs, 0, sizeof(qs)); + + poa_mgmt_msg_qos(&msg, &qs); + + if (memcmp(&qs, &qos_stream, sizeof(qs)) != 0) { + printf("QoS did not survive the management message.\n"); + goto fail; + } + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + fail: + TEST_FAIL(); + return TEST_RC_FAIL; +} + +#ifdef HAVE_RAW_SOCKETS + +static void eth_fake_priv(struct eth_priv * priv, + struct poa * e) +{ + memset(priv, 0, sizeof(*priv)); + memset(e, 0, sizeof(*e)); + + e->type = POA_ETH; + + strcpy(e->local.eth.src.dev, "test0"); + + priv->poa = e; + priv->ethertype = htons(0xA000); + priv->mtu = 1500; + + memcpy(priv->hw_addr, eth_our_mac, POA_MAC_SIZE); +} + +static int test_eth_frame(void) +{ + struct eth_priv priv; + struct poa e; + uint8_t buf[64]; + struct eth_hdr * hdr = (struct eth_hdr *) buf; + const char * body = "hello"; + size_t n; + size_t plen; + uint32_t eid; + + TEST_START(); + + eth_fake_priv(&priv, &e); + + eth_hdr_ser(&priv, hdr, eth_our_mac, 7, strlen(body)); + + memcpy(buf + ETH_HDR_TOT_SIZE, body, strlen(body)); + + n = ETH_HDR_TOT_SIZE + strlen(body); + if (frame_parse(&priv, buf, n, &eid, &plen) < 0) { + printf("Failed to parse a valid frame.\n"); + goto fail; + } + + if (eid != 7 || plen != strlen(body)) { + printf("Expected eid 7 len %zu, got %u len %zu.\n", + strlen(body), eid, plen); + goto fail; + } + + hdr->poa.hcs ^= 0xFF; + + if (frame_parse(&priv, buf, n, &eid, &plen) == 0) { + printf("Accepted a corrupt header checksum.\n"); + goto fail; + } + + hdr->poa.hcs ^= 0xFF; + hdr->ethertype ^= 0xFF; + + if (frame_parse(&priv, buf, n, &eid, &plen) == 0) { + printf("Accepted a foreign Ethertype.\n"); + goto fail; + } + + hdr->ethertype ^= 0xFF; + + if (frame_parse(&priv, buf, ETH_HDR_SIZE + 2, &eid, &plen) == 0) { + printf("Accepted a truncated frame.\n"); + goto fail; + } + + eth_hdr_ser(&priv, hdr, eth_far_mac, 7, strlen(body)); + + if (frame_parse(&priv, buf, n, &eid, &plen) == 0) { + printf("Accepted a frame for another host.\n"); + goto fail; + } + + eth_hdr_ser(&priv, hdr, eth_bc_mac, 7, strlen(body)); + + if (frame_parse(&priv, buf, n, &eid, &plen) < 0) { + printf("Rejected a broadcast frame.\n"); + goto fail; + } + + TEST_SUCCESS(); + + return TEST_RC_SUCCESS; + fail: + TEST_FAIL(); + return TEST_RC_FAIL; +} + +#endif /* HAVE_RAW_SOCKETS */ + +int poa_test(int argc, + char ** argv) +{ + int ret = 0; + + (void) argc; + (void) argv; + + ret |= test_poa_addr_cmp(); + ret |= test_poa_mgmt_msg_qos(); +#ifdef HAVE_RAW_SOCKETS + ret |= test_eth_frame(); +#endif + + return ret; +} |