From da60c56365ac13a262ffa6adaba7540c4d914843 Mon Sep 17 00:00:00 2001 From: Dimitri Staessens Date: Wed, 24 Oct 2018 10:06:23 +0200 Subject: ipcpd: Add broadcast IPCP This adds a broadcast IPCP that allows us to easily create multicast applications. The broadcast IPCP accepts flows for ".mc". A tool, obc (Ouroboros broadcast), is added that sends and reads a message to a broadcast layer. Signed-off-by: Dimitri Staessens Signed-off-by: Sander Vrijders --- src/ipcpd/broadcast/CMakeLists.txt | 34 +++ src/ipcpd/broadcast/comp.h | 45 ++++ src/ipcpd/broadcast/connmgr.c | 515 +++++++++++++++++++++++++++++++++++++ src/ipcpd/broadcast/connmgr.h | 71 +++++ src/ipcpd/broadcast/dt.c | 318 +++++++++++++++++++++++ src/ipcpd/broadcast/dt.h | 27 ++ src/ipcpd/broadcast/enroll.c | 363 ++++++++++++++++++++++++++ src/ipcpd/broadcast/enroll.h | 47 ++++ src/ipcpd/broadcast/main.c | 359 ++++++++++++++++++++++++++ 9 files changed, 1779 insertions(+) create mode 100644 src/ipcpd/broadcast/CMakeLists.txt create mode 100644 src/ipcpd/broadcast/comp.h create mode 100644 src/ipcpd/broadcast/connmgr.c create mode 100644 src/ipcpd/broadcast/connmgr.h create mode 100644 src/ipcpd/broadcast/dt.c create mode 100644 src/ipcpd/broadcast/dt.h create mode 100644 src/ipcpd/broadcast/enroll.c create mode 100644 src/ipcpd/broadcast/enroll.h create mode 100644 src/ipcpd/broadcast/main.c (limited to 'src/ipcpd/broadcast') diff --git a/src/ipcpd/broadcast/CMakeLists.txt b/src/ipcpd/broadcast/CMakeLists.txt new file mode 100644 index 00000000..afcc8696 --- /dev/null +++ b/src/ipcpd/broadcast/CMakeLists.txt @@ -0,0 +1,34 @@ +get_filename_component(CURRENT_SOURCE_PARENT_DIR + ${CMAKE_CURRENT_SOURCE_DIR} DIRECTORY) +get_filename_component(CURRENT_BINARY_PARENT_DIR + ${CMAKE_CURRENT_BINARY_DIR} DIRECTORY) + +include_directories(${CMAKE_CURRENT_SOURCE_DIR}) +include_directories(${CMAKE_CURRENT_BINARY_DIR}) + +include_directories(${CURRENT_SOURCE_PARENT_DIR}) +include_directories(${CURRENT_BINARY_PARENT_DIR}) + +include_directories(${CMAKE_SOURCE_DIR}/include) +include_directories(${CMAKE_BINARY_DIR}/include) + +set(IPCP_BROADCAST_TARGET ipcpd-broadcast CACHE INTERNAL "") + +set(SOURCE_FILES + # Add source files here + connmgr.c + dt.c + enroll.c + main.c + ) + +add_executable(ipcpd-broadcast ${SOURCE_FILES} ${IPCP_SOURCES} + ${LAYER_CONFIG_PROTO_SRCS}) +target_link_libraries(ipcpd-broadcast LINK_PUBLIC ouroboros-dev) + +include(AddCompileFlags) +if (CMAKE_BUILD_TYPE MATCHES "Debug*") + add_compile_flags(ipcpd-broadcast -DCONFIG_OUROBOROS_DEBUG) +endif () + +install(TARGETS ipcpd-broadcast RUNTIME DESTINATION ${CMAKE_INSTALL_SBINDIR}) diff --git a/src/ipcpd/broadcast/comp.h b/src/ipcpd/broadcast/comp.h new file mode 100644 index 00000000..a06f9a50 --- /dev/null +++ b/src/ipcpd/broadcast/comp.h @@ -0,0 +1,45 @@ +/* + * Ouroboros - Copyright (C) 2016 - 2018 + * + * Components for the broadcast IPC process + * + * Dimitri Staessens + * Sander Vrijders + * + * 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/. + */ + +#ifndef OUROBOROS_IPCPD_BROADCAST_COMP_H +#define OUROBOROS_IPCPD_BROADCAST_COMP_H + +#include + +#define DST_MAX_STRLEN 64 + +enum comp_id { + COMPID_DT = 0, + COMPID_ENROLL, + COMPID_MAX +}; + +struct conn { + struct conn_info conn_info; + struct { + char dst[DST_MAX_STRLEN + 1]; + int fd; + qosspec_t qs; + } flow_info; +}; + +#endif /* OUROBOROS_IPCPD_BROADCAST_COMP_H */ diff --git a/src/ipcpd/broadcast/connmgr.c b/src/ipcpd/broadcast/connmgr.c new file mode 100644 index 00000000..0398ba1f --- /dev/null +++ b/src/ipcpd/broadcast/connmgr.c @@ -0,0 +1,515 @@ +/* + * Ouroboros - Copyright (C) 2016 - 2018 + * + * Handles connections between components + * + * Dimitri Staessens + * Sander Vrijders + * + * 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 200112L +#endif + +#define OUROBOROS_PREFIX "connection-manager" + +#include +#include +#include +#include +#include +#include + +#include "comp.h" +#include "connmgr.h" +#include "enroll.h" +#include "ipcp.h" + +#include +#include +#include +#include + +enum connmgr_state { + CONNMGR_NULL = 0, + CONNMGR_INIT, + CONNMGR_RUNNING +}; + +struct conn_el { + struct list_head next; + struct conn conn; +}; + +struct comp { + struct conn_info info; + + struct list_head conns; + struct list_head pending; + + pthread_cond_t cond; + pthread_mutex_t lock; +}; + +struct { + struct comp comps[COMPID_MAX]; + enum connmgr_state state; + + pthread_t acceptor; +} connmgr; + +static int get_id_by_name(const char * name) +{ + enum comp_id i; + + for (i = 0; i < COMPID_MAX; ++i) + if (strcmp(name, connmgr.comps[i].info.comp_name) == 0) + return i; + + return -1; +} + +static int get_conn_by_fd(int fd, + enum comp_id id, + struct conn * conn) +{ + struct list_head * p; + + pthread_mutex_lock(&connmgr.comps[id].lock); + + list_for_each(p, &connmgr.comps[id].conns) { + struct conn_el * c = + list_entry(p, struct conn_el, next); + if (c->conn.flow_info.fd == fd) { + *conn = c->conn; + pthread_mutex_unlock(&connmgr.comps[id].lock); + return 0; + } + } + + pthread_mutex_unlock(&connmgr.comps[id].lock); + + return -1; +} + +static int add_comp_conn(enum comp_id id, + int fd, + qosspec_t qs, + struct conn_info * rcv_info) +{ + struct conn_el * el; + + el = malloc(sizeof(*el)); + if (el == NULL) { + log_err("Not enough memory."); + return -1; + } + + el->conn.conn_info = *rcv_info; + el->conn.flow_info.fd = fd; + el->conn.flow_info.qs = qs; + + pthread_mutex_lock(&connmgr.comps[id].lock); + + list_add(&el->next, &connmgr.comps[id].pending); + pthread_cond_signal(&connmgr.comps[id].cond); + + pthread_mutex_unlock(&connmgr.comps[id].lock); + + return 0; +} + +static void * flow_acceptor(void * o) +{ + int fd; + qosspec_t qs; + struct conn_info rcv_info; + struct conn_info fail_info; + + (void) o; + + memset(&fail_info, 0, sizeof(fail_info)); + + while (true) { + int id; + + fd = flow_accept(&qs, NULL); + if (fd < 0) { + if (fd != -EIRMD) + log_warn("Flow accept failed: %d", fd); + continue; + } + + if (cacep_rcv(fd, &rcv_info)) { + log_dbg("Error establishing application connection."); + flow_dealloc(fd); + continue; + } + + id = get_id_by_name(rcv_info.comp_name); + if (id < 0) { + log_dbg("Connection request for unknown component %s.", + rcv_info.comp_name); + cacep_snd(fd, &fail_info); + flow_dealloc(fd); + continue; + } + + assert(id < COMPID_MAX); + + if (cacep_snd(fd, &connmgr.comps[id].info)) { + log_dbg("Failed to respond to request."); + flow_dealloc(fd); + continue; + } + + if (add_comp_conn(id, fd, qs, &rcv_info)) { + log_dbg("Failed to add new connection."); + flow_dealloc(fd); + continue; + } + } + + return (void *) 0; +} + +static void handle_event(void * self, + int event, + const void * o) +{ + struct conn conn; + + (void) self; + + if (!(event == NOTIFY_DT_FLOW_UP || + event == NOTIFY_DT_FLOW_DOWN || + event == NOTIFY_DT_FLOW_DEALLOC)) + return; + + if (get_conn_by_fd(*((int *) o), COMPID_DT, &conn)) + return; + + switch (event) { + case NOTIFY_DT_FLOW_UP: + notifier_event(NOTIFY_DT_CONN_UP, &conn); + break; + case NOTIFY_DT_FLOW_DOWN: + notifier_event(NOTIFY_DT_CONN_DOWN, &conn); + break; + case NOTIFY_DT_FLOW_DEALLOC: + notifier_event(NOTIFY_DT_CONN_DEL, &conn); + break; + default: + break; + } +} + +int connmgr_init(void) +{ + connmgr.state = CONNMGR_INIT; + + if (notifier_reg(handle_event, NULL)) + return -1; + + return 0; +} + +void connmgr_fini(void) +{ + int i; + + notifier_unreg(handle_event); + + if (connmgr.state == CONNMGR_RUNNING) + pthread_join(connmgr.acceptor, NULL); + + for (i = 0; i < COMPID_MAX; ++i) + connmgr_comp_fini(i); +} + +int connmgr_start(void) +{ + if (pthread_create(&connmgr.acceptor, NULL, flow_acceptor, NULL)) + return -1; + + connmgr.state = CONNMGR_RUNNING; + + return 0; +} + +void connmgr_stop(void) +{ + if (connmgr.state == CONNMGR_RUNNING) + pthread_cancel(connmgr.acceptor); +} + +int connmgr_comp_init(enum comp_id id, + const struct conn_info * info) +{ + struct comp * comp; + + assert(id >= 0 && id < COMPID_MAX); + + comp = connmgr.comps + id; + + if (pthread_mutex_init(&comp->lock, NULL)) + return -1; + + if (pthread_cond_init(&comp->cond, NULL)) { + pthread_mutex_destroy(&comp->lock); + return -1; + } + + list_head_init(&comp->conns); + list_head_init(&comp->pending); + + memcpy(&connmgr.comps[id].info, info, sizeof(connmgr.comps[id].info)); + + return 0; +} + +void connmgr_comp_fini(enum comp_id id) +{ + struct list_head * p; + struct list_head * h; + struct comp * comp; + + assert(id >= 0 && id < COMPID_MAX); + + if (strlen(connmgr.comps[id].info.comp_name) == 0) + return; + + comp = connmgr.comps + id; + + pthread_mutex_lock(&comp->lock); + + list_for_each_safe(p, h, &comp->conns) { + struct conn_el * e = list_entry(p, struct conn_el, next); + list_del(&e->next); + free(e); + } + + list_for_each_safe(p, h, &comp->pending) { + struct conn_el * e = list_entry(p, struct conn_el, next); + list_del(&e->next); + free(e); + } + + pthread_mutex_unlock(&comp->lock); + + pthread_cond_destroy(&comp->cond); + pthread_mutex_destroy(&comp->lock); + + memset(&connmgr.comps[id].info, 0, sizeof(connmgr.comps[id].info)); +} + +int connmgr_ipcp_connect(const char * dst, + const char * component, + qosspec_t qs) +{ + struct conn_el * ce; + int id; + + assert(dst); + assert(component); + + ce = malloc(sizeof(*ce)); + if (ce == NULL) { + log_dbg("Out of memory."); + return -1; + } + + id = get_id_by_name(component); + if (id < 0) { + log_dbg("No such component: %s", component); + free(ce); + return -1; + } + + if (connmgr_alloc(id, dst, &qs, &ce->conn)) { + free(ce); + return -1; + } + + if (strlen(dst) > DST_MAX_STRLEN) { + log_warn("Truncating dst length for connection."); + memcpy(ce->conn.flow_info.dst, dst, DST_MAX_STRLEN); + ce->conn.flow_info.dst[DST_MAX_STRLEN] = '\0'; + } else { + strcpy(ce->conn.flow_info.dst, dst); + } + + pthread_mutex_lock(&connmgr.comps[id].lock); + + list_add(&ce->next, &connmgr.comps[id].conns); + + pthread_mutex_unlock(&connmgr.comps[id].lock); + + return 0; +} + +int connmgr_ipcp_disconnect(const char * dst, + const char * component) +{ + struct list_head * p; + struct list_head * h; + int id; + + assert(dst); + assert(component); + + id = get_id_by_name(component); + if (id < 0) + return -1; + + pthread_mutex_lock(&connmgr.comps[id].lock); + + list_for_each_safe(p,h, &connmgr.comps[id].conns) { + struct conn_el * el = list_entry(p, struct conn_el, next); + if (strcmp(el->conn.flow_info.dst, dst) == 0) { + int ret; + pthread_mutex_unlock(&connmgr.comps[id].lock); + list_del(&el->next); + ret = connmgr_dealloc(id, &el->conn); + free(el); + return ret; + } + } + + pthread_mutex_unlock(&connmgr.comps[id].lock); + + return 0; +} + +int connmgr_alloc(enum comp_id id, + const char * dst, + qosspec_t * qs, + struct conn * conn) +{ + assert(id >= 0 && id < COMPID_MAX); + assert(dst); + + conn->flow_info.fd = flow_alloc(dst, qs, NULL); + if (conn->flow_info.fd < 0) { + log_dbg("Failed to allocate flow to %s.", dst); + return -1; + } + + if (qs != NULL) + conn->flow_info.qs = *qs; + else + memset(&conn->flow_info.qs, 0, sizeof(conn->flow_info.qs)); + + log_dbg("Sending cacep info for protocol %s to fd %d.", + connmgr.comps[id].info.protocol, conn->flow_info.fd); + + if (cacep_snd(conn->flow_info.fd, &connmgr.comps[id].info)) { + log_dbg("Failed to create application connection."); + flow_dealloc(conn->flow_info.fd); + return -1; + } + + if (cacep_rcv(conn->flow_info.fd, &conn->conn_info)) { + log_dbg("Failed to connect to application."); + flow_dealloc(conn->flow_info.fd); + return -1; + } + + if (strcmp(connmgr.comps[id].info.protocol, conn->conn_info.protocol)) { + log_dbg("Unknown protocol (requested %s, got %s).", + connmgr.comps[id].info.protocol, + conn->conn_info.protocol); + flow_dealloc(conn->flow_info.fd); + return -1; + } + + if (connmgr.comps[id].info.pref_version != + conn->conn_info.pref_version) { + log_dbg("Unknown protocol version."); + flow_dealloc(conn->flow_info.fd); + return -1; + } + + if (connmgr.comps[id].info.pref_syntax != conn->conn_info.pref_syntax) { + log_dbg("Unknown protocol syntax."); + flow_dealloc(conn->flow_info.fd); + return -1; + } + + switch (id) { + case COMPID_DT: + notifier_event(NOTIFY_DT_CONN_ADD, conn); + break; + default: + break; + } + + return 0; +} + +int connmgr_dealloc(enum comp_id id, + struct conn * conn) +{ + switch (id) { + case COMPID_DT: + notifier_event(NOTIFY_DT_CONN_DEL, conn); + break; + default: + break; + } + + return flow_dealloc(conn->flow_info.fd); +} + + +int connmgr_wait(enum comp_id id, + struct conn * conn) +{ + struct conn_el * el; + struct comp * comp; + + assert(id >= 0 && id < COMPID_MAX); + assert(conn); + + comp = connmgr.comps + id; + + pthread_mutex_lock(&comp->lock); + + pthread_cleanup_push((void(*)(void *))pthread_mutex_unlock, + (void *) &comp->lock); + + while (list_is_empty(&comp->pending)) + pthread_cond_wait(&comp->cond, &comp->lock); + + pthread_cleanup_pop(false); + + el = list_first_entry((&comp->pending), struct conn_el, next); + if (el == NULL) { + pthread_mutex_unlock(&comp->lock); + return -1; + } + + *conn = el->conn; + + list_del(&el->next); + list_add(&el->next, &connmgr.comps[id].conns); + + pthread_mutex_unlock(&comp->lock); + + return 0; +} diff --git a/src/ipcpd/broadcast/connmgr.h b/src/ipcpd/broadcast/connmgr.h new file mode 100644 index 00000000..d1426613 --- /dev/null +++ b/src/ipcpd/broadcast/connmgr.h @@ -0,0 +1,71 @@ +/* + * Ouroboros - Copyright (C) 2016 - 2018 + * + * Handles the different AP connections + * + * Dimitri Staessens + * Sander Vrijders + * + * 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/. + */ + +#ifndef OUROBOROS_IPCPD_BROADCAST_CONNMGR_H +#define OUROBOROS_IPCPD_BROADCAST_CONNMGR_H + +#include +#include + +#include "comp.h" + +#define NOTIFY_DT_CONN_ADD 0x00D0 +#define NOTIFY_DT_CONN_DEL 0x00D1 +#define NOTIFY_DT_CONN_QOS 0x00D2 +#define NOTIFY_DT_CONN_UP 0x00D3 +#define NOTIFY_DT_CONN_DOWN 0x00D4 +#define NOTIFY_DT_FLOW_UP 0x00D5 +#define NOTIFY_DT_FLOW_DOWN 0x00D6 +#define NOTIFY_DT_FLOW_DEALLOC 0x00D7 + +int connmgr_init(void); + +void connmgr_fini(void); + +int connmgr_start(void); + +void connmgr_stop(void); + +int connmgr_comp_init(enum comp_id id, + const struct conn_info * info); + +void connmgr_comp_fini(enum comp_id id); + +int connmgr_ipcp_connect(const char * dst, + const char * component, + qosspec_t qs); + +int connmgr_ipcp_disconnect(const char * dst, + const char * component); + +int connmgr_alloc(enum comp_id id, + const char * dst, + qosspec_t * qs, + struct conn * conn); + +int connmgr_dealloc(enum comp_id id, + struct conn * conn); + +int connmgr_wait(enum comp_id id, + struct conn * conn); + +#endif /* OUROBOROS_IPCPD_BROADCAST_CONNMGR_H */ diff --git a/src/ipcpd/broadcast/dt.c b/src/ipcpd/broadcast/dt.c new file mode 100644 index 00000000..4d48ac84 --- /dev/null +++ b/src/ipcpd/broadcast/dt.c @@ -0,0 +1,318 @@ +/* + * Ouroboros - Copyright (C) 2016 - 2018 + * + * Forward loop for broadcast + * + * Dimitri Staessens + * Sander Vrijders + * + * 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 200112L +#endif + +#include "config.h" + +#define BROADCAST_MTU 1400 /* FIXME: avoid packet copy. */ + +#define DT "dt" +#define OUROBOROS_PREFIX DT + +#include +#include +#include +#include +#include +#include +#include +#include + +#include "comp.h" +#include "connmgr.h" +#include "dt.h" +#include "ipcp.h" + +#include +#include +#include +#include +#include + +struct nb { + struct list_head next; + + int fd; +}; + +struct { + struct list_head nbs; + size_t nbs_len; + pthread_rwlock_t nbs_lock; + + fset_t * set; + + pthread_t reader; + pthread_t listener; +} fwd; + +static int dt_add_nb(int fd) +{ + struct list_head * p; + struct nb * nb; + + pthread_rwlock_wrlock(&fwd.nbs_lock); + + list_for_each(p, &fwd.nbs) { + struct nb * el = list_entry(p, struct nb, next); + if (el->fd == fd) { + log_dbg("Already know neighbor."); + pthread_rwlock_unlock(&fwd.nbs_lock); + return -EPERM; + } + } + + nb = malloc(sizeof(*nb)); + if (nb == NULL) { + pthread_rwlock_unlock(&fwd.nbs_lock); + return -ENOMEM; + } + + nb->fd = fd; + + list_add_tail(&nb->next, p); + + ++fwd.nbs_len; + + log_dbg("Neighbor %d added.", fd); + + pthread_rwlock_unlock(&fwd.nbs_lock); + + return 0; +} + +static int dt_del_nb(int fd) +{ + struct list_head * p; + struct list_head * h; + + pthread_rwlock_wrlock(&fwd.nbs_lock); + + list_for_each_safe(p, h, &fwd.nbs) { + struct nb * nb = list_entry(p, struct nb, next); + if (nb->fd == fd) { + list_del(&nb->next); + --fwd.nbs_len; + pthread_rwlock_unlock(&fwd.nbs_lock); + log_dbg("Neighbor %d deleted.", nb->fd); + free(nb); + return 0; + } + } + + pthread_rwlock_unlock(&fwd.nbs_lock); + + return -EPERM; +} + +static void * dt_conn_handle(void * o) +{ + struct conn conn; + + (void) o; + + while (true) { + if (connmgr_wait(COMPID_DT, &conn)) { + log_err("Failed to get next DT connection."); + continue; + } + + /* NOTE: connection acceptance policy could be here. */ + + notifier_event(NOTIFY_DT_CONN_ADD, &conn); + } + + return 0; +} + + +static void dt_packet(uint8_t * buf, + size_t len, + int in_fd) +{ + struct list_head * p; + + pthread_rwlock_rdlock(&fwd.nbs_lock); + + pthread_cleanup_push((void (*))(void *) pthread_rwlock_unlock, + &fwd.nbs_lock); + + list_for_each(p, &fwd.nbs) { + struct nb * nb = list_entry(p, struct nb, next); + if (nb->fd != in_fd) + flow_write(nb->fd, buf, len); /* FIXME: avoid copy. */ + } + + pthread_cleanup_pop(true); +} + +static void * dt_reader(void * o) +{ + fqueue_t * fq; + int ret; + uint8_t buf[BROADCAST_MTU]; + int fd; + ssize_t len; + + (void) o; + + fq = fqueue_create(); + if (fq == NULL) + return (void *) -1; + + pthread_cleanup_push((void (*) (void *)) fqueue_destroy, + (void *) fq); + + while (true) { + ret = fevent(fwd.set, fq, NULL); + if (ret < 0) { + log_warn("Event error: %d.", ret); + continue; + } + + while ((fd = fqueue_next(fq)) >= 0) { + if (fqueue_type(fq) != FLOW_PKT) + continue; + + /* FIXME: avoid copy. */ + len = flow_read(fd, buf, BROADCAST_MTU); + if (len < 0) + continue; + + dt_packet(buf, len, fd); + } + } + + pthread_cleanup_pop(true); + + return (void *) 0; +} + +static void handle_event(void * self, + int event, + const void * o) +{ + /* FIXME: Apply correct QoS on graph */ + struct conn * c; + + (void) self; + + c = (struct conn *) o; + + switch (event) { + case NOTIFY_DT_CONN_ADD: + if (dt_add_nb(c->flow_info.fd)) + log_dbg("Failed to add neighbor."); + fset_add(fwd.set, c->flow_info.fd); + break; + case NOTIFY_DT_CONN_DEL: + if (dt_del_nb(c->flow_info.fd)) + log_dbg("Failed to delete neighbor."); + fset_del(fwd.set, c->flow_info.fd); + break; + default: + break; + } +} + +int dt_init(void) +{ + struct conn_info info; + + memset(&info, 0, sizeof(info)); + + strcpy(info.comp_name, DT); + strcpy(info.comp_name, DT_COMP); + + list_head_init(&fwd.nbs); + + if (notifier_reg(handle_event, NULL)) + goto fail_notifier_reg; + + if (pthread_rwlock_init(&fwd.nbs_lock, NULL)) + goto fail_lock_init; + + fwd.set = fset_create(); + if (fwd.set == NULL) + goto fail_fset_create; + + if (pthread_create(&fwd.reader, NULL, dt_reader, NULL)) + goto fail_pthread_create_reader; + + if (pthread_create(&fwd.listener, NULL, dt_conn_handle, NULL)) + goto fail_pthread_create_listener; + + if (connmgr_comp_init(COMPID_DT, &info)) + goto fail_connmgr_comp_init; + + fwd.nbs_len = 0; + + return 0; + + fail_connmgr_comp_init: + pthread_cancel(fwd.listener); + pthread_join(fwd.listener, NULL); + fail_pthread_create_listener: + pthread_cancel(fwd.reader); + pthread_join(fwd.reader, NULL); + fail_pthread_create_reader: + fset_destroy(fwd.set); + fail_fset_create: + pthread_rwlock_destroy(&fwd.nbs_lock); + fail_lock_init: + notifier_unreg(handle_event); + fail_notifier_reg: + return -1; +} + +void dt_fini(void) +{ + struct list_head * p; + struct list_head * h; + + notifier_unreg(handle_event); + + pthread_cancel(fwd.reader); + pthread_cancel(fwd.listener); + + pthread_join(fwd.reader, NULL); + pthread_join(fwd.listener, NULL); + + fset_destroy(fwd.set); + + pthread_rwlock_wrlock(&fwd.nbs_lock); + + list_for_each_safe(p, h, &fwd.nbs) { + struct nb * n = list_entry(p, struct nb, next); + list_del(&n->next); + free(n); + } + + pthread_rwlock_unlock(&fwd.nbs_lock); + + pthread_rwlock_destroy(&fwd.nbs_lock); +} diff --git a/src/ipcpd/broadcast/dt.h b/src/ipcpd/broadcast/dt.h new file mode 100644 index 00000000..853ce03e --- /dev/null +++ b/src/ipcpd/broadcast/dt.h @@ -0,0 +1,27 @@ +/* + * Ouroboros - Copyright (C) 2016 - 2018 + * + * Forward loop for broadcast + * + * Dimitri Staessens + * Sander Vrijders + * + * 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/. + */ + +#define DT_COMP "Data Transfer" + +int dt_init(void); + +void dt_fini(void); diff --git a/src/ipcpd/broadcast/enroll.c b/src/ipcpd/broadcast/enroll.c new file mode 100644 index 00000000..f8641d31 --- /dev/null +++ b/src/ipcpd/broadcast/enroll.c @@ -0,0 +1,363 @@ +/* + * Ouroboros - Copyright (C) 2016 - 2018 + * + * Enrollment Task + * + * Dimitri Staessens + * Sander Vrijders + * + * 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 199309L +#endif + +#define OUROBOROS_PREFIX "enrollment" + +#include +#include +#include +#include +#include +#include +#include + +#include "connmgr.h" +#include "enroll.h" +#include "ipcp.h" + +#include +#include +#include +#include + +#include "ipcp_config.pb-c.h" +typedef EnrollMsg enroll_msg_t; + +#define ENROLL_COMP "Enrollment" +#define ENROLL_PROTO "OEP" /* Ouroboros enrollment protocol */ +#define ENROLL_WARN_TIME_OFFSET 20 +#define ENROLL_BUF_LEN 1024 + +enum enroll_state { + ENROLL_NULL = 0, + ENROLL_INIT, + ENROLL_RUNNING +}; + +struct { + struct ipcp_config conf; + enum enroll_state state; + pthread_t listener; +} enroll; + +static int send_rcv_enroll_msg(int fd) +{ + enroll_msg_t req = ENROLL_MSG__INIT; + enroll_msg_t * reply; + uint8_t buf[ENROLL_BUF_LEN]; + ssize_t len; + ssize_t delta_t; + struct timespec t0; + struct timespec rtt; + + req.code = ENROLL_CODE__ENROLL_REQ; + + len = enroll_msg__get_packed_size(&req); + if (len < 0) { + log_dbg("Failed pack request message."); + return -1; + } + + enroll_msg__pack(&req, buf); + + clock_gettime(CLOCK_REALTIME, &t0); + + if (flow_write(fd, buf, len)) { + log_dbg("Failed to send request message."); + return -1; + } + + len = flow_read(fd, buf, ENROLL_BUF_LEN); + if (len < 0) { + log_dbg("No enrollment reply received."); + return -1; + } + + log_dbg("Received enrollment info (%zd bytes).", len); + + reply = enroll_msg__unpack(NULL, len, buf); + if (reply == NULL) { + log_dbg("No enrollment response."); + return -1; + } + + if (reply->code != ENROLL_CODE__ENROLL_BOOT) { + log_dbg("Failed to unpack enrollment response."); + enroll_msg__free_unpacked(reply, NULL); + return -1; + } + + if (!(reply->has_t_sec && reply->has_t_nsec)) { + log_dbg("No time in response message."); + enroll_msg__free_unpacked(reply, NULL); + return -1; + } + + clock_gettime(CLOCK_REALTIME, &rtt); + + delta_t = ts_diff_ms(&t0, &rtt); + + rtt.tv_sec = reply->t_sec; + rtt.tv_nsec = reply->t_nsec; + + if (labs(ts_diff_ms(&t0, &rtt)) - delta_t > ENROLL_WARN_TIME_OFFSET) + log_warn("Clock offset above threshold."); + + strcpy(enroll.conf.layer_info.layer_name, + reply->conf->layer_info->layer_name); + enroll.conf.layer_info.dir_hash_algo + = reply->conf->layer_info->dir_hash_algo; + + enroll_msg__free_unpacked(reply, NULL); + + return 0; +} + +static ssize_t enroll_pack(uint8_t ** buf) +{ + enroll_msg_t msg = ENROLL_MSG__INIT; + ipcp_config_msg_t config = IPCP_CONFIG_MSG__INIT; + layer_info_msg_t layer_info = LAYER_INFO_MSG__INIT; + struct timespec now; + ssize_t len; + + clock_gettime(CLOCK_REALTIME, &now); + + msg.code = ENROLL_CODE__ENROLL_BOOT; + msg.has_t_sec = true; + msg.t_sec = now.tv_sec; + msg.has_t_nsec = true; + msg.t_nsec = now.tv_nsec; + msg.conf = &config; + + config.ipcp_type = enroll.conf.type; + config.layer_info = &layer_info; + + layer_info.layer_name = (char *) enroll.conf.layer_info.layer_name; + layer_info.dir_hash_algo = enroll.conf.layer_info.dir_hash_algo; + + len = enroll_msg__get_packed_size(&msg); + + *buf = malloc(len); + if (*buf == NULL) + return -1; + + enroll_msg__pack(&msg, *buf); + + return len; +} + +static void * enroll_handle(void * o) +{ + struct conn conn; + uint8_t buf[ENROLL_BUF_LEN]; + uint8_t * reply; + ssize_t len; + enroll_msg_t * msg; + + (void) o; + + while (true) { + if (connmgr_wait(COMPID_ENROLL, &conn)) { + log_err("Failed to get next connection."); + continue; + } + + len = flow_read(conn.flow_info.fd, buf, ENROLL_BUF_LEN); + if (len < 0) { + log_err("Failed to read from flow."); + connmgr_dealloc(COMPID_ENROLL, &conn); + continue; + } + + msg = enroll_msg__unpack(NULL, len, buf); + if (msg == NULL) { + log_err("Failed to unpack message."); + connmgr_dealloc(COMPID_ENROLL, &conn); + continue; + } + + if (msg->code != ENROLL_CODE__ENROLL_REQ) { + log_err("Wrong message type."); + connmgr_dealloc(COMPID_ENROLL, &conn); + enroll_msg__free_unpacked(msg, NULL); + continue; + } + + log_dbg("Enrolling a new neighbor."); + + enroll_msg__free_unpacked(msg, NULL); + + len = enroll_pack(&reply); + if (reply == NULL) { + log_err("Failed to pack enrollment message."); + connmgr_dealloc(COMPID_ENROLL, &conn); + continue; + } + + log_dbg("Sending enrollment info (%zd bytes).", len); + + if (flow_write(conn.flow_info.fd, reply, len)) { + log_err("Failed respond to enrollment request."); + connmgr_dealloc(COMPID_ENROLL, &conn); + free(reply); + continue; + } + + free(reply); + + len = flow_read(conn.flow_info.fd, buf, ENROLL_BUF_LEN); + if (len < 0) { + log_err("Failed to read from flow."); + connmgr_dealloc(COMPID_ENROLL, &conn); + continue; + } + + msg = enroll_msg__unpack(NULL, len, buf); + if (msg == NULL) { + log_err("Failed to unpack message."); + connmgr_dealloc(COMPID_ENROLL, &conn); + continue; + } + + if (msg->code != ENROLL_CODE__ENROLL_DONE || !msg->has_result) { + log_err("Wrong message type."); + enroll_msg__free_unpacked(msg, NULL); + connmgr_dealloc(COMPID_ENROLL, &conn); + continue; + } + + if (msg->result == 0) + log_dbg("Neighbor enrollment successful."); + else + log_dbg("Neigbor reported failed enrollment."); + + enroll_msg__free_unpacked(msg, NULL); + + connmgr_dealloc(COMPID_ENROLL, &conn); + } + + return 0; +} + +int enroll_boot(struct conn * conn) +{ + log_dbg("Getting boot information."); + + if (send_rcv_enroll_msg(conn->flow_info.fd)) { + log_err("Failed to enroll."); + return -1; + } + + return 0; +} + +int enroll_done(struct conn * conn, + int result) +{ + enroll_msg_t msg = ENROLL_MSG__INIT; + uint8_t buf[ENROLL_BUF_LEN]; + ssize_t len; + + msg.code = ENROLL_CODE__ENROLL_DONE; + msg.has_result = true; + msg.result = result; + + len = enroll_msg__get_packed_size(&msg); + if (len < 0) { + log_dbg("Failed pack request message."); + return -1; + } + + enroll_msg__pack(&msg, buf); + + if (flow_write(conn->flow_info.fd, buf, len)) { + log_dbg("Failed to send acknowledgment."); + return -1; + } + + return 0; +} + +void enroll_bootstrap(const struct ipcp_config * conf) +{ + assert(conf); + + memcpy(&enroll.conf, conf, sizeof(enroll.conf)); +} + +struct ipcp_config * enroll_get_conf(void) +{ + return &enroll.conf; +} + +int enroll_init(void) +{ + struct conn_info info; + + memset(&info, 0, sizeof(info)); + + strcpy(info.comp_name, ENROLL_COMP); + strcpy(info.protocol, ENROLL_PROTO); + info.pref_version = 1; + info.pref_syntax = PROTO_GPB; + info.addr = 0; + + if (connmgr_comp_init(COMPID_ENROLL, &info)) { + log_err("Failed to register with connmgr."); + return -1; + } + + enroll.state = ENROLL_INIT; + + return 0; +} + +void enroll_fini(void) +{ + if (enroll.state == ENROLL_RUNNING) + pthread_join(enroll.listener, NULL); + + connmgr_comp_fini(COMPID_ENROLL); +} + +int enroll_start(void) +{ + if (pthread_create(&enroll.listener, NULL, enroll_handle, NULL)) + return -1; + + enroll.state = ENROLL_RUNNING; + + return 0; +} + +void enroll_stop(void) +{ + if (enroll.state == ENROLL_RUNNING) + pthread_cancel(enroll.listener); +} diff --git a/src/ipcpd/broadcast/enroll.h b/src/ipcpd/broadcast/enroll.h new file mode 100644 index 00000000..728c3ad2 --- /dev/null +++ b/src/ipcpd/broadcast/enroll.h @@ -0,0 +1,47 @@ +/* + * Ouroboros - Copyright (C) 2016 - 2018 + * + * Enrollment Task + * + * Dimitri Staessens + * Sander Vrijders + * + * 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/. + */ + +#ifndef OUROBOROS_IPCPD_BROADCAST_ENROLL_H +#define OUROBOROS_IPCPD_BROADCAST_ENROLL_H + +#include + +#include "comp.h" + +int enroll_init(void); + +void enroll_fini(void); + +int enroll_start(void); + +void enroll_stop(void); + +void enroll_bootstrap(const struct ipcp_config * conf); + +int enroll_boot(struct conn * conn); + +int enroll_done(struct conn * conn, + int result); + +struct ipcp_config * enroll_get_conf(void); + +#endif /* OUROBOROS_IPCPD_BROADCAST_ENROLL_H */ diff --git a/src/ipcpd/broadcast/main.c b/src/ipcpd/broadcast/main.c new file mode 100644 index 00000000..8c6bfa71 --- /dev/null +++ b/src/ipcpd/broadcast/main.c @@ -0,0 +1,359 @@ +/* + * Ouroboros - Copyright (C) 2016 - 2018 + * + * Broadcast IPC Process + * + * Dimitri Staessens + * Sander Vrijders + * + * 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 "config.h" + +#define OUROBOROS_PREFIX "broadcast-ipcp" + +#include +#include +#include +#include +#include +#include +#include +#include + +#include "connmgr.h" +#include "dt.h" +#include "enroll.h" +#include "ipcp.h" + +#include +#include +#include +#include +#include +#include + +#define THIS_TYPE IPCP_BROADCAST + +static int initialize_components(const struct ipcp_config * conf) +{ + ipcpi.layer_name = strdup(conf->layer_info.layer_name); + if (ipcpi.layer_name == NULL) { + log_err("Failed to set layer name."); + goto fail_layer_name; + } + + ipcpi.dir_hash_algo = conf->layer_info.dir_hash_algo; + + assert(ipcp_dir_hash_len() != 0); + + if (dt_init()) { + log_err("Failed to initialize forwarding component."); + goto fail_dt; + } + + ipcp_set_state(IPCP_INIT); + + return 0; + + fail_dt: + free(ipcpi.layer_name); + fail_layer_name: + return -1; +} + +static void finalize_components(void) +{ + dt_fini(); + + free(ipcpi.layer_name); +} + +static int start_components(void) +{ + assert(ipcp_get_state() == IPCP_INIT); + + ipcp_set_state(IPCP_OPERATIONAL); + + if (enroll_start()) { + log_err("Failed to start enrollment."); + goto fail_enroll_start; + } + + if (connmgr_start()) { + log_err("Failed to start AP connection manager."); + goto fail_connmgr_start; + } + + return 0; + + fail_connmgr_start: + enroll_stop(); + fail_enroll_start: + ipcp_set_state(IPCP_INIT); + return -1; +} + +static void stop_components(void) +{ + assert(ipcp_get_state() == IPCP_OPERATIONAL || + ipcp_get_state() == IPCP_SHUTDOWN); + + connmgr_stop(); + + enroll_stop(); + + ipcp_set_state(IPCP_INIT); +} + +static int broadcast_ipcp_enroll(const char * dst, + struct layer_info * info) +{ + struct conn conn; + + if (connmgr_alloc(COMPID_ENROLL, dst, NULL, &conn)) { + log_err("Failed to get connection."); + goto fail_er_flow; + } + + /* Get boot state from peer. */ + if (enroll_boot(&conn)) { + log_err("Failed to get boot information."); + goto fail_enroll_boot; + } + + if (initialize_components(enroll_get_conf())) { + log_err("Failed to initialize IPCP components."); + goto fail_enroll_boot; + } + + if (start_components()) { + log_err("Failed to start components."); + goto fail_start_comp; + } + + if (enroll_done(&conn, 0)) + log_warn("Failed to confirm enrollment with peer."); + + if (connmgr_dealloc(COMPID_ENROLL, &conn)) + log_warn("Failed to deallocate enrollment flow."); + + log_info("Enrolled with %s.", dst); + + info->dir_hash_algo = ipcpi.dir_hash_algo; + strcpy(info->layer_name, ipcpi.layer_name); + + return 0; + + fail_start_comp: + finalize_components(); + fail_enroll_boot: + connmgr_dealloc(COMPID_ENROLL, &conn); + fail_er_flow: + return -1; +} + +static int broadcast_ipcp_bootstrap(const struct ipcp_config * conf) +{ + assert(conf); + assert(conf->type == THIS_TYPE); + + enroll_bootstrap(conf); + + if (initialize_components(conf)) { + log_err("Failed to init IPCP components."); + goto fail_init; + } + + if (start_components()) { + log_err("Failed to init IPCP components."); + goto fail_start; + } + + log_dbg("Bootstrapped in layer %s.", conf->layer_info.layer_name); + + return 0; + + fail_start: + finalize_components(); + fail_init: + return -1; +} + +static int broadcast_ipcp_query(const uint8_t * dst) +{ + uint8_t * buf; + size_t len; + int ret; + char * multicast_name; + char * suffix = ".mc"; + + len = hash_len(ipcpi.dir_hash_algo); + buf = malloc(len); + if (buf == NULL) + return -ENOMEM; + + multicast_name = malloc(strlen(ipcpi.layer_name) + strlen(suffix) + 1); + if (multicast_name == NULL) { + free(buf); + return -ENOMEM; + } + + strcpy(multicast_name, ipcpi.layer_name); + strcat(multicast_name, suffix); + + str_hash(ipcpi.dir_hash_algo, buf, multicast_name); + + free(multicast_name); + + ret = memcmp(buf, dst, len); + + free(buf); + + return ret; +} + +static int broadcast_ipcp_alloc(int fd, + const uint8_t * dst, + qosspec_t qs) +{ + struct conn conn; + + (void) qs; + + memset(&conn, 0, sizeof(conn)); + + conn.flow_info.fd = fd; + + if (broadcast_ipcp_query(dst) != 0) + return -1; + + notifier_event(NOTIFY_DT_CONN_ADD, &conn); + + ipcp_flow_alloc_reply(fd, 0); + + return 0; +} + +int broadcast_ipcp_dealloc(int fd) +{ + struct conn conn; + + memset(&conn, 0, sizeof(conn)); + + conn.flow_info.fd = fd; + + notifier_event(NOTIFY_DT_CONN_DEL, &conn); + + flow_dealloc(fd); + + return 0; +} + + +static struct ipcp_ops broadcast_ops = { + .ipcp_bootstrap = broadcast_ipcp_bootstrap, + .ipcp_enroll = broadcast_ipcp_enroll, + .ipcp_connect = connmgr_ipcp_connect, + .ipcp_disconnect = connmgr_ipcp_disconnect, + .ipcp_reg = NULL, + .ipcp_unreg = NULL, + .ipcp_query = broadcast_ipcp_query, + .ipcp_flow_alloc = broadcast_ipcp_alloc, + .ipcp_flow_alloc_resp = NULL, + .ipcp_flow_dealloc = broadcast_ipcp_dealloc +}; + +int main(int argc, + char * argv[]) +{ + if (ipcp_init(argc, argv, &broadcast_ops) < 0) { + log_err("Failed to init IPCP."); + goto fail_init; + } + + /* These components must be init at creation. */ + if (rib_init(ipcpi.name)) { + log_err("Failed to initialize RIB."); + goto fail_rib_init; + } + + if (notifier_init()) { + log_err("Failed to initialize notifier component."); + goto fail_notifier_init; + } + + if (connmgr_init()) { + log_err("Failed to initialize connection manager."); + goto fail_connmgr_init; + } + + if (enroll_init()) { + log_err("Failed to initialize enrollment component."); + goto fail_enroll_init; + } + + if (ipcp_boot() < 0) { + log_err("Failed to boot IPCP."); + goto fail_boot; + } + + if (ipcp_create_r(getpid(), 0)) { + log_err("Failed to notify IRMd we are initialized."); + ipcp_set_state(IPCP_NULL); + goto fail_create_r; + } + + ipcp_shutdown(); + + if (ipcp_get_state() == IPCP_SHUTDOWN) { + stop_components(); + finalize_components(); + } + + enroll_fini(); + + connmgr_fini(); + + notifier_fini(); + + rib_fini(); + + ipcp_fini(); + + exit(EXIT_SUCCESS); + + fail_create_r: + ipcp_shutdown(); + fail_boot: + enroll_fini(); + fail_enroll_init: + connmgr_fini(); + fail_connmgr_init: + notifier_fini(); + fail_notifier_init: + rib_fini(); + fail_rib_init: + ipcp_fini(); + fail_init: + ipcp_create_r(getpid(), -1); + exit(EXIT_FAILURE); +} -- cgit v1.2.3