X-Git-Url: http://git.meshlink.io/?a=blobdiff_plain;f=src%2Fmeshlink.c;h=98e02f7db623d8c8b4e837eb96e77f23580fddd8;hb=5678b294c5bbb61b983a7a269bc85fd6ea6590ff;hp=2c1253c79864d10b0cf567b233ffce4523a88191;hpb=5aa1e39a0ff8304e24151d3a69a68737084269e2;p=meshlink diff --git a/src/meshlink.c b/src/meshlink.c index 2c1253c7..98e02f7d 100644 --- a/src/meshlink.c +++ b/src/meshlink.c @@ -333,7 +333,7 @@ void remove_duplicate_hostnames(char *host[], char *port[], int n) { break; } - if(found) { + if(found || !is_valid_hostname(host[i])) { free(host[i]); free(port[i]); host[i] = NULL; @@ -417,7 +417,7 @@ static char *get_my_hostname(meshlink_handle_t *mesh, uint32_t flags) { char resolved_port[NI_MAXSERV]; err = getnameinfo(ai_in->ai_addr, ai_in->ai_addrlen, resolved_host, sizeof resolved_host, resolved_port, sizeof resolved_port, NI_NUMERICSERV); - if(err) { + if(err || !is_valid_hostname(resolved_host)) { freeaddrinfo(ai_in); continue; } @@ -514,9 +514,9 @@ static bool try_bind(int port) { return true; } -int check_port(meshlink_handle_t *mesh) { +static int check_port(meshlink_handle_t *mesh) { for(int i = 0; i < 1000; i++) { - int port = 0x1000 + (rand() & 0x7fff); + int port = 0x1000 + prng(mesh, 0x8000); if(try_bind(port)) { free(mesh->myport); @@ -575,7 +575,7 @@ static bool finalize_join(meshlink_handle_t *mesh, const void *buf, uint16_t len char *name = packmsg_get_str_dup(&in); packmsg_skip_element(&in); /* submesh */ - int32_t devclass = packmsg_get_int32(&in); + dev_class_t devclass = packmsg_get_int32(&in); uint32_t count = packmsg_get_array(&in); if(!name) { @@ -599,7 +599,7 @@ static bool finalize_join(meshlink_handle_t *mesh, const void *buf, uint16_t len free(mesh->self->name); mesh->name = name; mesh->self->name = xstrdup(name); - mesh->self->devclass = devclass; + mesh->self->devclass = devclass == DEV_CLASS_UNKNOWN ? mesh->devclass : devclass; // Initialize configuration directory if(!config_init(mesh, "current")) { @@ -661,7 +661,12 @@ static bool finalize_join(meshlink_handle_t *mesh, const void *buf, uint16_t len } } - sptps_send_record(&(mesh->sptps), 1, ecdsa_get_public_key(mesh->private_key), 32); + /* Ensure the configuration directory metadata is on disk */ + if(!config_sync(mesh, "current")) { + return false; + } + + sptps_send_record(&mesh->sptps, 1, ecdsa_get_public_key(mesh->private_key), 32); logger(mesh, MESHLINK_DEBUG, "Configuration stored in: %s\n", mesh->confbase); @@ -694,7 +699,7 @@ static bool invitation_receive(void *handle, uint8_t type, const void *msg, uint switch(type) { case SPTPS_HANDSHAKE: - return sptps_send_record(&(mesh->sptps), 0, mesh->cookie, sizeof(mesh)->cookie); + return sptps_send_record(&mesh->sptps, 0, mesh->cookie, sizeof(mesh)->cookie); case 0: return finalize_join(mesh, msg, len); @@ -744,8 +749,9 @@ static bool recvline(meshlink_handle_t *mesh, size_t len) { return true; } + static bool sendline(int fd, char *format, ...) { - static char buffer[4096]; + char buffer[4096]; char *p = buffer; int blen = 0; va_list ap; @@ -790,6 +796,7 @@ static const char *errstr[] = { [MESHLINK_EPEER] = "Error communicating with peer", [MESHLINK_ENOTSUP] = "Operation not supported", [MESHLINK_EBUSY] = "MeshLink instance already in use", + [MESHLINK_EBLACKLISTED] = "Node is blacklisted", }; const char *meshlink_strerror(meshlink_errno_t err) { @@ -840,6 +847,7 @@ static struct timeval idle(event_loop_t *loop, void *data) { // Get our local address(es) by simulating connecting to an Internet host. static void add_local_addresses(meshlink_handle_t *mesh) { struct sockaddr_storage sn; + sn.ss_family = AF_UNKNOWN; socklen_t sl = sizeof(sn); // IPv4 example.org @@ -882,6 +890,7 @@ static bool meshlink_setup(meshlink_handle_t *mesh) { mesh->self->name = xstrdup(mesh->name); mesh->self->devclass = mesh->devclass; mesh->self->ecdsa = ecdsa_set_public_key(ecdsa_get_public_key(mesh->private_key)); + mesh->self->session_id = mesh->session_id; if(!write_main_config_files(mesh)) { logger(mesh, MESHLINK_ERROR, "Could not write main config files into %s/current: %s\n", mesh->confbase, strerror(errno)); @@ -889,6 +898,11 @@ static bool meshlink_setup(meshlink_handle_t *mesh) { return false; } + /* Ensure the configuration directory metadata is on disk */ + if(!config_sync(mesh, "current")) { + return false; + } + if(!main_config_lock(mesh)) { logger(NULL, MESHLINK_ERROR, "Cannot lock main config file\n"); meshlink_errno = MESHLINK_ESTORAGE; @@ -955,6 +969,7 @@ static bool meshlink_read_config(meshlink_handle_t *mesh) { mesh->self = new_node(); mesh->self->name = xstrdup(name); mesh->self->devclass = mesh->devclass; + mesh->self->session_id = mesh->session_id; if(!node_read_public_key(mesh, mesh->self)) { logger(NULL, MESHLINK_ERROR, "Could not read our host configuration file!"); @@ -1002,16 +1017,17 @@ meshlink_open_params_t *meshlink_open_params_init(const char *confbase, const ch if(!name || !*name) { logger(NULL, MESHLINK_ERROR, "No name given!\n"); - //return NULL; - } else { //check name only if there is a name != NULL - if(!check_id(name)) { - logger(NULL, MESHLINK_ERROR, "Invalid name given!\n"); - meshlink_errno = MESHLINK_EINVAL; - return NULL; - } + meshlink_errno = MESHLINK_EINVAL; + return NULL; + }; + + if(!check_id(name)) { + logger(NULL, MESHLINK_ERROR, "Invalid name given!\n"); + meshlink_errno = MESHLINK_EINVAL; + return NULL; } - if((int)devclass < 0 || devclass > _DEV_CLASS_MAX) { + if(devclass < 0 || devclass >= DEV_CLASS_COUNT) { logger(NULL, MESHLINK_ERROR, "Invalid devclass given!\n"); meshlink_errno = MESHLINK_EINVAL; return NULL; @@ -1064,7 +1080,7 @@ bool meshlink_encrypted_key_rotate(meshlink_handle_t *mesh, const void *new_key, return false; } - pthread_mutex_lock(&(mesh->mesh_mutex)); + pthread_mutex_lock(&mesh->mutex); // Create hash for the new key void *new_config_key; @@ -1073,7 +1089,7 @@ bool meshlink_encrypted_key_rotate(meshlink_handle_t *mesh, const void *new_key, if(!prf(new_key, new_keylen, "MeshLink configuration key", 26, new_config_key, CHACHA_POLY1305_KEYLEN)) { logger(mesh, MESHLINK_ERROR, "Error creating new configuration key!\n"); meshlink_errno = MESHLINK_EINTERNAL; - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return false; } @@ -1082,7 +1098,7 @@ bool meshlink_encrypted_key_rotate(meshlink_handle_t *mesh, const void *new_key, if(!config_copy(mesh, "current", mesh->config_key, "new", new_config_key)) { logger(mesh, MESHLINK_ERROR, "Could not set up configuration in %s/old: %s\n", mesh->confbase, strerror(errno)); meshlink_errno = MESHLINK_ESTORAGE; - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return false; } @@ -1096,7 +1112,7 @@ bool meshlink_encrypted_key_rotate(meshlink_handle_t *mesh, const void *new_key, logger(mesh, MESHLINK_ERROR, "Cannot rename %s/current to %s/old\n", mesh->confbase, mesh->confbase); meshlink_errno = MESHLINK_ESTORAGE; main_config_lock(mesh); - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return false; } @@ -1108,21 +1124,21 @@ bool meshlink_encrypted_key_rotate(meshlink_handle_t *mesh, const void *new_key, logger(mesh, MESHLINK_ERROR, "Cannot rename %s/new to %s/current\n", mesh->confbase, mesh->confbase); meshlink_errno = MESHLINK_ESTORAGE; main_config_lock(mesh); - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return false; } devtool_keyrotate_probe(3); if(!main_config_lock(mesh)) { - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return false; } // Cleanup the "old" confbase sub-directory if(!config_destroy(mesh->confbase, "old")) { - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return false; } @@ -1131,7 +1147,7 @@ bool meshlink_encrypted_key_rotate(meshlink_handle_t *mesh, const void *new_key, free(mesh->config_key); mesh->config_key = new_config_key; - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return true; } @@ -1149,6 +1165,14 @@ void meshlink_open_params_free(meshlink_open_params_t *params) { free(params); } +/// Device class traits +static const dev_class_traits_t default_class_traits[DEV_CLASS_COUNT] = { + { .pingtimeout = 5, .pinginterval = 60, .min_connects = 3, .max_connects = 10000, .edge_weight = 1 }, // DEV_CLASS_BACKBONE + { .pingtimeout = 5, .pinginterval = 60, .min_connects = 3, .max_connects = 100, .edge_weight = 3 }, // DEV_CLASS_STATIONARY + { .pingtimeout = 5, .pinginterval = 60, .min_connects = 3, .max_connects = 3, .edge_weight = 6 }, // DEV_CLASS_PORTABLE + { .pingtimeout = 5, .pinginterval = 60, .min_connects = 1, .max_connects = 1, .edge_weight = 9 }, // DEV_CLASS_UNKNOWN +}; + meshlink_handle_t *meshlink_open(const char *confbase, const char *name, const char *appname, dev_class_t devclass) { if(!confbase || !*confbase) { logger(NULL, MESHLINK_ERROR, "No confbase given!\n"); @@ -1177,7 +1201,8 @@ meshlink_handle_t *meshlink_open_encrypted(const char *confbase, const char *nam } /* Create a temporary struct on the stack, to avoid allocating and freeing one. */ - meshlink_open_params_t params = {NULL}; + meshlink_open_params_t params; + memset(¶ms, 0, sizeof(params)); params.confbase = (char *)confbase; params.name = (char *)name; @@ -1194,7 +1219,8 @@ meshlink_handle_t *meshlink_open_encrypted(const char *confbase, const char *nam meshlink_handle_t *meshlink_open_ephemeral(const char *name, const char *appname, dev_class_t devclass) { /* Create a temporary struct on the stack, to avoid allocating and freeing one. */ - meshlink_open_params_t params = {NULL}; + meshlink_open_params_t params; + memset(¶ms, 0, sizeof(params)); params.name = (char *)name; params.appname = (char *)appname; @@ -1236,7 +1262,7 @@ meshlink_handle_t *meshlink_open_ex(const meshlink_open_params_t *params) { } } - if((int)params->devclass < 0 || params->devclass > _DEV_CLASS_MAX) { + if(params->devclass < 0 || params->devclass >= DEV_CLASS_COUNT) { logger(NULL, MESHLINK_ERROR, "Invalid devclass given!\n"); meshlink_errno = MESHLINK_EINVAL; return NULL; @@ -1260,6 +1286,16 @@ meshlink_handle_t *meshlink_open_ex(const meshlink_open_params_t *params) { mesh->invitation_timeout = 604800; // 1 week mesh->netns = params->netns; mesh->submeshes = NULL; + mesh->log_cb = global_log_cb; + mesh->log_level = global_log_level; + + randomize(&mesh->prng_state, sizeof(mesh->prng_state)); + + do { + randomize(&mesh->session_id, sizeof(mesh->session_id)); + } while(mesh->session_id == 0); + + memcpy(mesh->dev_class_traits, default_class_traits, sizeof(default_class_traits)); if(usingname) { mesh->name = xstrdup(params->name); @@ -1281,7 +1317,7 @@ meshlink_handle_t *meshlink_open_ex(const meshlink_open_params_t *params) { pthread_mutexattr_t attr; pthread_mutexattr_init(&attr); pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_RECURSIVE); - pthread_mutex_init(&(mesh->mesh_mutex), &attr); + pthread_mutex_init(&mesh->mutex, &attr); mesh->threadstarted = false; event_loop_init(&mesh->loop); @@ -1365,11 +1401,11 @@ meshlink_submesh_t *meshlink_submesh_open(meshlink_handle_t *mesh, const char * } //lock mesh->nodes - pthread_mutex_lock(&(mesh->mesh_mutex)); + pthread_mutex_lock(&mesh->mutex); s = (meshlink_submesh_t *)create_submesh(mesh, submesh); - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return s; } @@ -1381,16 +1417,16 @@ static void *meshlink_main_loop(void *arg) { #ifdef HAVE_SETNS if(setns(mesh->netns, CLONE_NEWNET) != 0) { + pthread_cond_signal(&mesh->cond); return NULL; } #else + pthread_cond_signal(&mesh->cond); return NULL; #endif // HAVE_SETNS } - pthread_mutex_lock(&(mesh->mesh_mutex)); - #if HAVE_CATTA if(mesh->discovery) { @@ -1399,10 +1435,15 @@ static void *meshlink_main_loop(void *arg) { #endif + pthread_mutex_lock(&mesh->mutex); + logger(mesh, MESHLINK_DEBUG, "Starting main_loop...\n"); + pthread_cond_broadcast(&mesh->cond); main_loop(mesh); logger(mesh, MESHLINK_DEBUG, "main_loop returned.\n"); + pthread_mutex_unlock(&mesh->mutex); + #if HAVE_CATTA // Stop discovery @@ -1412,7 +1453,6 @@ static void *meshlink_main_loop(void *arg) { #endif - pthread_mutex_unlock(&(mesh->mesh_mutex)); return NULL; } @@ -1427,11 +1467,14 @@ bool meshlink_start(meshlink_handle_t *mesh) { logger(mesh, MESHLINK_DEBUG, "meshlink_start called\n"); - pthread_mutex_lock(&(mesh->mesh_mutex)); + pthread_mutex_lock(&mesh->mutex); + + assert(mesh->self->ecdsa); + assert(!memcmp((uint8_t *)mesh->self->ecdsa + 64, (uint8_t *)mesh->private_key + 64, 32)); if(mesh->threadstarted) { logger(mesh, MESHLINK_DEBUG, "thread was already running\n"); - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return true; } @@ -1449,7 +1492,7 @@ bool meshlink_start(meshlink_handle_t *mesh) { if(!mesh->name) { logger(mesh, MESHLINK_DEBUG, "No name given!\n"); meshlink_errno = MESHLINK_EINVAL; - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return false; } @@ -1464,17 +1507,14 @@ bool meshlink_start(meshlink_handle_t *mesh) { memset(&mesh->thread, 0, sizeof(mesh)->thread); meshlink_errno = MESHLINK_EINTERNAL; event_loop_stop(&mesh->loop); - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return false; } + pthread_cond_wait(&mesh->cond, &mesh->mutex); mesh->threadstarted = true; - assert(mesh->self->ecdsa); - assert(!memcmp((uint8_t *)mesh->self->ecdsa + 64, (uint8_t *)mesh->private_key + 64, 32)); - - - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return true; } @@ -1484,7 +1524,7 @@ void meshlink_stop(meshlink_handle_t *mesh) { return; } - pthread_mutex_lock(&(mesh->mesh_mutex)); + pthread_mutex_lock(&mesh->mutex); logger(mesh, MESHLINK_DEBUG, "meshlink_stop called\n"); // Shut down the main thread @@ -1507,9 +1547,9 @@ void meshlink_stop(meshlink_handle_t *mesh) { if(mesh->threadstarted) { // Wait for the main thread to finish - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); pthread_join(mesh->thread, NULL); - pthread_mutex_lock(&(mesh->mesh_mutex)); + pthread_mutex_lock(&mesh->mutex); mesh->threadstarted = false; } @@ -1536,7 +1576,7 @@ void meshlink_stop(meshlink_handle_t *mesh) { } } - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); } void meshlink_close(meshlink_handle_t *mesh) { @@ -1549,7 +1589,7 @@ void meshlink_close(meshlink_handle_t *mesh) { meshlink_stop(mesh); // lock is not released after this - pthread_mutex_lock(&(mesh->mesh_mutex)); + pthread_mutex_lock(&mesh->mutex); // Close and free all resources used. @@ -1573,15 +1613,23 @@ void meshlink_close(meshlink_handle_t *mesh) { close(mesh->netns); } + for(vpn_packet_t *packet; (packet = meshlink_queue_pop(&mesh->outpacketqueue));) { + free(packet); + } + + meshlink_queue_exit(&mesh->outpacketqueue); + free(mesh->name); free(mesh->appname); free(mesh->confbase); free(mesh->config_key); ecdsa_free(mesh->private_key); - pthread_mutex_destroy(&(mesh->mesh_mutex)); main_config_unlock(mesh); + pthread_mutex_unlock(&mesh->mutex); + pthread_mutex_destroy(&mesh->mutex); + memset(mesh, 0, sizeof(*mesh)); free(mesh); @@ -1616,9 +1664,9 @@ void meshlink_set_receive_cb(meshlink_handle_t *mesh, meshlink_receive_cb_t cb) return; } - pthread_mutex_lock(&(mesh->mesh_mutex)); + pthread_mutex_lock(&mesh->mutex); mesh->receive_cb = cb; - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); } void meshlink_set_connection_try_cb(meshlink_handle_t *mesh, meshlink_connection_try_cb_t cb) { @@ -1627,9 +1675,9 @@ void meshlink_set_connection_try_cb(meshlink_handle_t *mesh, meshlink_connection return; } - pthread_mutex_lock(&(mesh->mesh_mutex)); + pthread_mutex_lock(&mesh->mutex); mesh->connection_try_cb = cb; - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); } void meshlink_set_node_status_cb(meshlink_handle_t *mesh, meshlink_node_status_cb_t cb) { @@ -1638,9 +1686,20 @@ void meshlink_set_node_status_cb(meshlink_handle_t *mesh, meshlink_node_status_c return; } - pthread_mutex_lock(&(mesh->mesh_mutex)); + pthread_mutex_lock(&mesh->mutex); mesh->node_status_cb = cb; - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); +} + +void meshlink_set_node_pmtu_cb(meshlink_handle_t *mesh, meshlink_node_pmtu_cb_t cb) { + if(!mesh) { + meshlink_errno = MESHLINK_EINVAL; + return; + } + + pthread_mutex_lock(&mesh->mutex); + mesh->node_pmtu_cb = cb; + pthread_mutex_unlock(&mesh->mutex); } void meshlink_set_node_duplicate_cb(meshlink_handle_t *mesh, meshlink_node_duplicate_cb_t cb) { @@ -1649,23 +1708,34 @@ void meshlink_set_node_duplicate_cb(meshlink_handle_t *mesh, meshlink_node_dupli return; } - pthread_mutex_lock(&(mesh->mesh_mutex)); + pthread_mutex_lock(&mesh->mutex); mesh->node_duplicate_cb = cb; - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); } void meshlink_set_log_cb(meshlink_handle_t *mesh, meshlink_log_level_t level, meshlink_log_cb_t cb) { if(mesh) { - pthread_mutex_lock(&(mesh->mesh_mutex)); + pthread_mutex_lock(&mesh->mutex); mesh->log_cb = cb; mesh->log_level = cb ? level : 0; - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); } else { global_log_cb = cb; global_log_level = cb ? level : 0; } } +void meshlink_set_error_cb(struct meshlink_handle *mesh, meshlink_error_cb_t cb) { + if(!mesh) { + meshlink_errno = MESHLINK_EINVAL; + return; + } + + pthread_mutex_lock(&mesh->mutex); + mesh->error_cb = cb; + pthread_mutex_unlock(&mesh->mutex); +} + bool meshlink_send(meshlink_handle_t *mesh, meshlink_node_t *destination, const void *data, size_t len) { meshlink_packethdr_t *hdr; @@ -1688,6 +1758,7 @@ bool meshlink_send(meshlink_handle_t *mesh, meshlink_node_t *destination, const if(n->status.blacklisted) { logger(mesh, MESHLINK_ERROR, "Node %s blacklisted, dropping packet\n", n->name); + meshlink_errno = MESHLINK_EBLACKLISTED; return false; } @@ -1720,13 +1791,14 @@ bool meshlink_send(meshlink_handle_t *mesh, meshlink_node_t *destination, const } // Notify event loop - signal_trigger(&(mesh->loop), &(mesh->datafromapp)); + signal_trigger(&mesh->loop, &mesh->datafromapp); return true; } -void meshlink_send_from_queue(event_loop_t *loop, meshlink_handle_t *mesh) { +void meshlink_send_from_queue(event_loop_t *loop, void *data) { (void)loop; + meshlink_handle_t *mesh = data; vpn_packet_t *packet = meshlink_queue_pop(&mesh->outpacketqueue); if(!packet) { @@ -1736,6 +1808,8 @@ void meshlink_send_from_queue(event_loop_t *loop, meshlink_handle_t *mesh) { mesh->self->in_packets++; mesh->self->in_bytes += packet->len; route(mesh, mesh->self, packet); + + free(packet); } ssize_t meshlink_get_pmtu(meshlink_handle_t *mesh, meshlink_node_t *destination) { @@ -1744,19 +1818,19 @@ ssize_t meshlink_get_pmtu(meshlink_handle_t *mesh, meshlink_node_t *destination) return -1; } - pthread_mutex_lock(&(mesh->mesh_mutex)); + pthread_mutex_lock(&mesh->mutex); node_t *n = (node_t *)destination; if(!n->status.reachable) { - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return 0; } else if(n->mtuprobes > 30 && n->minmtu) { - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return n->minmtu; } else { - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return MTU; } } @@ -1767,13 +1841,13 @@ char *meshlink_get_fingerprint(meshlink_handle_t *mesh, meshlink_node_t *node) { return NULL; } - pthread_mutex_lock(&(mesh->mesh_mutex)); + pthread_mutex_lock(&mesh->mutex); node_t *n = (node_t *)node; if(!node_read_public_key(mesh, n) || !n->ecdsa) { meshlink_errno = MESHLINK_EINTERNAL; - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return false; } @@ -1783,7 +1857,7 @@ char *meshlink_get_fingerprint(meshlink_handle_t *mesh, meshlink_node_t *node) { meshlink_errno = MESHLINK_EINTERNAL; } - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return fingerprint; } @@ -1804,9 +1878,14 @@ meshlink_node_t *meshlink_get_node(meshlink_handle_t *mesh, const char *name) { meshlink_node_t *node = NULL; - pthread_mutex_lock(&(mesh->mesh_mutex)); + pthread_mutex_lock(&mesh->mutex); node = (meshlink_node_t *)lookup_node(mesh, (char *)name); // TODO: make lookup_node() use const - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); + + if(!node) { + meshlink_errno = MESHLINK_ENOENT; + } + return node; } @@ -1818,9 +1897,14 @@ meshlink_submesh_t *meshlink_get_submesh(meshlink_handle_t *mesh, const char *na meshlink_submesh_t *submesh = NULL; - pthread_mutex_lock(&(mesh->mesh_mutex)); + pthread_mutex_lock(&mesh->mutex); submesh = (meshlink_submesh_t *)lookup_submesh(mesh, name); - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); + + if(!submesh) { + meshlink_errno = MESHLINK_ENOENT; + } + return submesh; } @@ -1833,7 +1917,7 @@ meshlink_node_t **meshlink_get_all_nodes(meshlink_handle_t *mesh, meshlink_node_ meshlink_node_t **result; //lock mesh->nodes - pthread_mutex_lock(&(mesh->mesh_mutex)); + pthread_mutex_lock(&mesh->mutex); *nmemb = mesh->nodes->count; result = realloc(nodes, *nmemb * sizeof(*nodes)); @@ -1850,7 +1934,7 @@ meshlink_node_t **meshlink_get_all_nodes(meshlink_handle_t *mesh, meshlink_node_ meshlink_errno = MESHLINK_ENOMEM; } - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return result; } @@ -1858,7 +1942,7 @@ meshlink_node_t **meshlink_get_all_nodes(meshlink_handle_t *mesh, meshlink_node_ static meshlink_node_t **meshlink_get_all_nodes_by_condition(meshlink_handle_t *mesh, const void *condition, meshlink_node_t **nodes, size_t *nmemb, search_node_by_condition_t search_node) { meshlink_node_t **result; - pthread_mutex_lock(&(mesh->mesh_mutex)); + pthread_mutex_lock(&mesh->mutex); *nmemb = 0; @@ -1870,7 +1954,7 @@ static meshlink_node_t **meshlink_get_all_nodes_by_condition(meshlink_handle_t * if(*nmemb == 0) { free(nodes); - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return NULL; } @@ -1890,7 +1974,7 @@ static meshlink_node_t **meshlink_get_all_nodes_by_condition(meshlink_handle_t * meshlink_errno = MESHLINK_ENOMEM; } - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return result; } @@ -1914,7 +1998,7 @@ static bool search_node_by_submesh(const node_t *node, const void *condition) { } meshlink_node_t **meshlink_get_all_nodes_by_dev_class(meshlink_handle_t *mesh, dev_class_t devclass, meshlink_node_t **nodes, size_t *nmemb) { - if(!mesh || ((int)devclass < 0) || (devclass > _DEV_CLASS_MAX) || !nmemb) { + if(!mesh || devclass < 0 || devclass >= DEV_CLASS_COUNT || !nmemb) { meshlink_errno = MESHLINK_EINVAL; return NULL; } @@ -1939,11 +2023,11 @@ dev_class_t meshlink_get_node_dev_class(meshlink_handle_t *mesh, meshlink_node_t dev_class_t devclass; - pthread_mutex_lock(&(mesh->mesh_mutex)); + pthread_mutex_lock(&mesh->mutex); devclass = ((node_t *)node)->devclass; - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return devclass; } @@ -1974,16 +2058,16 @@ bool meshlink_sign(meshlink_handle_t *mesh, const void *data, size_t len, void * return false; } - pthread_mutex_lock(&(mesh->mesh_mutex)); + pthread_mutex_lock(&mesh->mutex); if(!ecdsa_sign(mesh->private_key, data, len, signature)) { meshlink_errno = MESHLINK_EINTERNAL; - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return false; } *siglen = MESHLINK_SIGLEN; - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return true; } @@ -1998,7 +2082,7 @@ bool meshlink_verify(meshlink_handle_t *mesh, meshlink_node_t *source, const voi return false; } - pthread_mutex_lock(&(mesh->mesh_mutex)); + pthread_mutex_lock(&mesh->mutex); bool rval = false; @@ -2011,12 +2095,12 @@ bool meshlink_verify(meshlink_handle_t *mesh, meshlink_node_t *source, const voi rval = ecdsa_verify(((struct node_t *)source)->ecdsa, data, len, signature); } - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return rval; } static bool refresh_invitation_key(meshlink_handle_t *mesh) { - pthread_mutex_lock(&(mesh->mesh_mutex)); + pthread_mutex_lock(&mesh->mutex); size_t count = invitation_purge_old(mesh, time(NULL) - mesh->invitation_timeout); @@ -2024,7 +2108,7 @@ static bool refresh_invitation_key(meshlink_handle_t *mesh) { // TODO: Update invitation key if necessary? } - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return mesh->invitation_key; } @@ -2055,14 +2139,14 @@ bool meshlink_set_canonical_address(meshlink_handle_t *mesh, meshlink_node_t *no canonical_address = xstrdup(address); } - pthread_mutex_lock(&(mesh->mesh_mutex)); + pthread_mutex_lock(&mesh->mutex); node_t *n = (node_t *)node; free(n->canonical_address); n->canonical_address = canonical_address; node_write_config(mesh, n); - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return true; } @@ -2100,7 +2184,13 @@ int meshlink_get_port(meshlink_handle_t *mesh) { return -1; } - return atoi(mesh->myport); + int port; + + pthread_mutex_lock(&mesh->mutex); + port = atoi(mesh->myport); + pthread_mutex_unlock(&mesh->mutex); + + return port; } bool meshlink_set_port(meshlink_handle_t *mesh, int port) { @@ -2122,7 +2212,7 @@ bool meshlink_set_port(meshlink_handle_t *mesh, int port) { bool rval = false; - pthread_mutex_lock(&(mesh->mesh_mutex)); + pthread_mutex_lock(&mesh->mutex); if(mesh->threadstarted) { meshlink_errno = MESHLINK_EINVAL; @@ -2132,9 +2222,6 @@ bool meshlink_set_port(meshlink_handle_t *mesh, int port) { free(mesh->myport); xasprintf(&mesh->myport, "%d", port); - /* Write meshlink.conf with the updated port number */ - write_main_config_files(mesh); - /* Close down the network. This also deletes mesh->self. */ close_network_connections(mesh); @@ -2142,6 +2229,7 @@ bool meshlink_set_port(meshlink_handle_t *mesh, int port) { mesh->self = new_node(); mesh->self->name = xstrdup(mesh->name); mesh->self->devclass = mesh->devclass; + mesh->self->session_id = mesh->session_id; xasprintf(&mesh->myport, "%d", port); if(!node_read_public_key(mesh, mesh->self)) { @@ -2149,14 +2237,23 @@ bool meshlink_set_port(meshlink_handle_t *mesh, int port) { meshlink_errno = MESHLINK_ESTORAGE; free_node(mesh->self); mesh->self = NULL; + goto done; } else if(!setup_network(mesh)) { meshlink_errno = MESHLINK_ENETWORK; - } else { - rval = true; + goto done; } + /* Rebuild our own list of recent addresses */ + memset(mesh->self->recent, 0, sizeof(mesh->self->recent)); + add_local_addresses(mesh); + + /* Write meshlink.conf with the updated port number */ + write_main_config_files(mesh); + + rval = config_sync(mesh, "current"); + done: - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return rval && meshlink_get_port(mesh) == port; } @@ -2185,13 +2282,13 @@ char *meshlink_invite_ex(meshlink_handle_t *mesh, meshlink_submesh_t *submesh, c s = (meshlink_submesh_t *)mesh->self->submesh; } - pthread_mutex_lock(&(mesh->mesh_mutex)); + pthread_mutex_lock(&mesh->mutex); // Check validity of the new node's name if(!check_id(name)) { logger(mesh, MESHLINK_DEBUG, "Invalid name for node.\n"); meshlink_errno = MESHLINK_EINVAL; - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return NULL; } @@ -2199,7 +2296,7 @@ char *meshlink_invite_ex(meshlink_handle_t *mesh, meshlink_submesh_t *submesh, c if(config_exists(mesh, "current", name)) { logger(mesh, MESHLINK_DEBUG, "A host config file for %s already exists!\n", name); meshlink_errno = MESHLINK_EEXIST; - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return NULL; } @@ -2207,7 +2304,7 @@ char *meshlink_invite_ex(meshlink_handle_t *mesh, meshlink_submesh_t *submesh, c if(meshlink_get_node(mesh, name)) { logger(mesh, MESHLINK_DEBUG, "A node with name %s is already known!\n", name); meshlink_errno = MESHLINK_EEXIST; - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return NULL; } @@ -2217,13 +2314,13 @@ char *meshlink_invite_ex(meshlink_handle_t *mesh, meshlink_submesh_t *submesh, c if(!address) { logger(mesh, MESHLINK_DEBUG, "No Address known for ourselves!\n"); meshlink_errno = MESHLINK_ERESOLV; - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return NULL; } if(!refresh_invitation_key(mesh)) { meshlink_errno = MESHLINK_EINTERNAL; - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return NULL; } @@ -2263,7 +2360,8 @@ char *meshlink_invite_ex(meshlink_handle_t *mesh, meshlink_submesh_t *submesh, c * Note: make sure we only add config files of nodes that are in the core mesh or the same submesh, * and are not blacklisted. */ - config_t configs[5] = {NULL}; + config_t configs[5]; + memset(configs, 0, sizeof(configs)); int count = 0; if(config_read(mesh, "current", mesh->self->name, &configs[count], mesh->config_key)) { @@ -2283,7 +2381,7 @@ char *meshlink_invite_ex(meshlink_handle_t *mesh, meshlink_submesh_t *submesh, c if(!invitation_write(mesh, "current", cookiehash, &config, mesh->config_key)) { logger(mesh, MESHLINK_DEBUG, "Could not create invitation file %s: %s\n", cookiehash, strerror(errno)); meshlink_errno = MESHLINK_ESTORAGE; - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return NULL; } @@ -2292,7 +2390,7 @@ char *meshlink_invite_ex(meshlink_handle_t *mesh, meshlink_submesh_t *submesh, c xasprintf(&url, "%s/%s%s", address, hash, cookie); free(address); - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return url; } @@ -2306,13 +2404,13 @@ bool meshlink_join(meshlink_handle_t *mesh, const char *invitation) { return false; } - pthread_mutex_lock(&(mesh->mesh_mutex)); + pthread_mutex_lock(&mesh->mutex); //Before doing meshlink_join make sure we are not connected to another mesh if(mesh->threadstarted) { logger(mesh, MESHLINK_ERROR, "Cannot join while started\n"); meshlink_errno = MESHLINK_EINVAL; - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return false; } @@ -2320,7 +2418,7 @@ bool meshlink_join(meshlink_handle_t *mesh, const char *invitation) { if(mesh->nodes->count > 1) { logger(mesh, MESHLINK_ERROR, "Already part of an existing mesh\n"); meshlink_errno = MESHLINK_EINVAL; - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return false; } @@ -2354,7 +2452,7 @@ bool meshlink_join(meshlink_handle_t *mesh, const char *invitation) { if(!key) { meshlink_errno = MESHLINK_EINTERNAL; - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return false; } @@ -2432,7 +2530,7 @@ bool meshlink_join(meshlink_handle_t *mesh, const char *invitation) { } if(mesh->sock == -1) { - pthread_mutex_unlock(&mesh->mesh_mutex); + pthread_mutex_unlock(&mesh->mutex); return false; } @@ -2446,7 +2544,7 @@ bool meshlink_join(meshlink_handle_t *mesh, const char *invitation) { logger(mesh, MESHLINK_DEBUG, "Error sending request to %s port %s: %s\n", address, port, strerror(errno)); closesocket(mesh->sock); meshlink_errno = MESHLINK_ENETWORK; - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return false; } @@ -2459,7 +2557,7 @@ bool meshlink_join(meshlink_handle_t *mesh, const char *invitation) { logger(mesh, MESHLINK_DEBUG, "Cannot read greeting from peer\n"); closesocket(mesh->sock); meshlink_errno = MESHLINK_ENETWORK; - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return false; } @@ -2470,14 +2568,14 @@ bool meshlink_join(meshlink_handle_t *mesh, const char *invitation) { if(sha512(fingerprint, strlen(fingerprint), hishash)) { logger(mesh, MESHLINK_DEBUG, "Could not create hash\n%s\n", mesh->line + 2); meshlink_errno = MESHLINK_EINTERNAL; - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return false; } if(memcmp(hishash, mesh->hash, 18)) { logger(mesh, MESHLINK_DEBUG, "Peer has an invalid key!\n%s\n", mesh->line + 2); meshlink_errno = MESHLINK_EPEER; - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return false; } @@ -2486,21 +2584,21 @@ bool meshlink_join(meshlink_handle_t *mesh, const char *invitation) { if(!hiskey) { meshlink_errno = MESHLINK_EINTERNAL; - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return false; } // Start an SPTPS session if(!sptps_start(&mesh->sptps, mesh, true, false, key, hiskey, meshlink_invitation_label, sizeof(meshlink_invitation_label), invitation_send, invitation_receive)) { meshlink_errno = MESHLINK_EINTERNAL; - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return false; } // Feed rest of input buffer to SPTPS if(!sptps_receive_data(&mesh->sptps, mesh->buffer, mesh->blen)) { meshlink_errno = MESHLINK_EPEER; - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return false; } @@ -2514,13 +2612,13 @@ bool meshlink_join(meshlink_handle_t *mesh, const char *invitation) { logger(mesh, MESHLINK_DEBUG, "Error reading data from %s port %s: %s\n", address, port, strerror(errno)); meshlink_errno = MESHLINK_ENETWORK; - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return false; } if(!sptps_receive_data(&mesh->sptps, mesh->line, len)) { meshlink_errno = MESHLINK_EPEER; - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return false; } } @@ -2533,17 +2631,17 @@ bool meshlink_join(meshlink_handle_t *mesh, const char *invitation) { if(!mesh->success) { logger(mesh, MESHLINK_DEBUG, "Connection closed by peer, invitation cancelled.\n"); meshlink_errno = MESHLINK_EPEER; - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return false; } - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return true; invalid: logger(mesh, MESHLINK_DEBUG, "Invalid invitation URL\n"); meshlink_errno = MESHLINK_EINVAL; - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return false; } @@ -2561,7 +2659,7 @@ char *meshlink_export(meshlink_handle_t *mesh) { packmsg_add_str(&out, mesh->name); packmsg_add_str(&out, CORE_MESH); - pthread_mutex_lock(&(mesh->mesh_mutex)); + pthread_mutex_lock(&mesh->mutex); packmsg_add_int32(&out, mesh->self->devclass); packmsg_add_bool(&out, mesh->self->status.blacklisted); @@ -2584,7 +2682,7 @@ char *meshlink_export(meshlink_handle_t *mesh) { packmsg_add_sockaddr(&out, &mesh->self->recent[i]); } - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); if(!packmsg_output_ok(&out)) { logger(mesh, MESHLINK_DEBUG, "Error creating export data\n"); @@ -2638,7 +2736,7 @@ bool meshlink_import(meshlink_handle_t *mesh, const char *data) { return false; } - pthread_mutex_lock(&(mesh->mesh_mutex)); + pthread_mutex_lock(&mesh->mutex); while(count--) { const void *data; @@ -2686,7 +2784,9 @@ bool meshlink_import(meshlink_handle_t *mesh, const char *data) { node_add(mesh, n); } - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); + + free(buf); if(!packmsg_done(&in)) { logger(mesh, MESHLINK_ERROR, "Invalid data\n"); @@ -2694,6 +2794,10 @@ bool meshlink_import(meshlink_handle_t *mesh, const char *data) { return false; } + if(!config_sync(mesh, "current")) { + return false; + } + return true; } @@ -2703,7 +2807,7 @@ void meshlink_blacklist(meshlink_handle_t *mesh, meshlink_node_t *node) { return; } - pthread_mutex_lock(&(mesh->mesh_mutex)); + pthread_mutex_lock(&mesh->mutex); node_t *n; n = (node_t *)node; @@ -2711,18 +2815,20 @@ void meshlink_blacklist(meshlink_handle_t *mesh, meshlink_node_t *node) { if(n == mesh->self) { logger(mesh, MESHLINK_ERROR, "%s blacklisting itself?\n", node->name); meshlink_errno = MESHLINK_EINVAL; - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return; } if(n->status.blacklisted) { logger(mesh, MESHLINK_DEBUG, "Node %s already blacklisted\n", node->name); - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return; } n->status.blacklisted = true; node_write_config(mesh, n); + config_sync(mesh, "current"); + logger(mesh, MESHLINK_DEBUG, "Blacklisted %s.\n", node->name); //Immediately terminate any connections we have with the blacklisted node @@ -2744,7 +2850,7 @@ void meshlink_blacklist(meshlink_handle_t *mesh, meshlink_node_t *node) { update_node_status(mesh, n); } - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); } void meshlink_whitelist(meshlink_handle_t *mesh, meshlink_node_t *node) { @@ -2753,21 +2859,34 @@ void meshlink_whitelist(meshlink_handle_t *mesh, meshlink_node_t *node) { return; } - pthread_mutex_lock(&(mesh->mesh_mutex)); + pthread_mutex_lock(&mesh->mutex); node_t *n = (node_t *)node; + if(n == mesh->self) { + logger(mesh, MESHLINK_ERROR, "%s whitelisting itself?\n", node->name); + meshlink_errno = MESHLINK_EINVAL; + pthread_mutex_unlock(&mesh->mutex); + return; + } + if(!n->status.blacklisted) { logger(mesh, MESHLINK_DEBUG, "Node %s was already whitelisted\n", node->name); - meshlink_errno = MESHLINK_EINVAL; - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return; } n->status.blacklisted = false; node_write_config(mesh, n); + config_sync(mesh, "current"); + + if(n->status.reachable) { + update_node_status(mesh, n); + } + + logger(mesh, MESHLINK_DEBUG, "Whitelisted %s.\n", node->name); - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); return; } @@ -2784,14 +2903,14 @@ void meshlink_hint_address(meshlink_handle_t *mesh, meshlink_node_t *node, const return; } - pthread_mutex_lock(&(mesh->mesh_mutex)); + pthread_mutex_lock(&mesh->mutex); node_t *n = (node_t *)node; memmove(n->recent + 1, n->recent, 4 * sizeof(*n->recent)); memcpy(n->recent, addr, SALEN(*addr)); node_write_config(mesh, n); - pthread_mutex_unlock(&(mesh->mesh_mutex)); + pthread_mutex_unlock(&mesh->mutex); // @TODO do we want to fire off a connection attempt right away? } @@ -2802,6 +2921,18 @@ static bool channel_pre_accept(struct utcp *utcp, uint16_t port) { return mesh->channel_accept_cb; } +static void aio_signal(meshlink_handle_t *mesh, meshlink_channel_t *channel, meshlink_aio_buffer_t *aio) { + if(aio->data) { + if(aio->cb.buffer) { + aio->cb.buffer(mesh, channel, aio->data, aio->len, aio->priv); + } + } else { + if(aio->cb.fd) { + aio->cb.fd(mesh, channel, aio->fd, aio->done, aio->priv); + } + } +} + static ssize_t channel_recv(struct utcp_connection *connection, const void *data, size_t len) { meshlink_channel_t *channel = connection->priv; @@ -2814,8 +2945,48 @@ static ssize_t channel_recv(struct utcp_connection *connection, const void *data if(n->status.destroyed) { meshlink_channel_close(mesh, channel); - } else if(channel->receive_cb) { - channel->receive_cb(mesh, channel, data, len); + return len; + } + + const char *p = data; + size_t left = len; + + while(channel->aio_receive) { + meshlink_aio_buffer_t *aio = channel->aio_receive; + size_t todo = aio->len - aio->done; + + if(todo > left) { + todo = left; + } + + if(aio->data) { + memcpy((char *)aio->data + aio->done, p, todo); + } else { + ssize_t result = write(aio->fd, p, todo); + + if(result > 0) { + todo = result; + } + } + + aio->done += todo; + + if(aio->done == aio->len) { + channel->aio_receive = aio->next; + aio_signal(mesh, channel, aio); + free(aio); + } + + p += todo; + left -= todo; + + if(!left && len) { + return len; + } + } + + if(channel->receive_cb) { + channel->receive_cb(mesh, channel, p, left); } return len; @@ -2885,16 +3056,69 @@ static void channel_poll(struct utcp_connection *connection, size_t len) { node_t *n = channel->node; meshlink_handle_t *mesh = n->mesh; + meshlink_aio_buffer_t *aio = channel->aio_send; + + if(aio) { + /* We at least one AIO buffer. Send as much as possible form the first buffer. */ + size_t left = aio->len - aio->done; + ssize_t sent; + + if(len > left) { + len = left; + } + + if(aio->data) { + sent = utcp_send(connection, (char *)aio->data + aio->done, len); + } else { + char buf[65536]; + size_t todo = utcp_get_sndbuf_free(connection); + + if(todo > left) { + todo = left; + } + + if(todo > sizeof(buf)) { + todo = sizeof(buf); + } + + ssize_t result = read(aio->fd, buf, todo); + + if(result > 0) { + sent = utcp_send(connection, buf, result); + } else { + sent = result; + } + } - if(channel->poll_cb) { - channel->poll_cb(mesh, channel, len); + if(sent >= 0) { + aio->done += sent; + } + + /* If the buffer is now completely sent, call the callback and dispose of it. */ + if(aio->done >= aio->len) { + channel->aio_send = aio->next; + aio_signal(mesh, channel, aio); + free(aio); + } + } else { + if(channel->poll_cb) { + channel->poll_cb(mesh, channel, len); + } else { + utcp_set_poll_cb(connection, NULL); + } } } void meshlink_set_channel_poll_cb(meshlink_handle_t *mesh, meshlink_channel_t *channel, meshlink_channel_poll_cb_t cb) { - (void)mesh; + if(!mesh || !channel) { + meshlink_errno = MESHLINK_EINVAL; + return; + } + + pthread_mutex_lock(&mesh->mutex); channel->poll_cb = cb; - utcp_set_poll_cb(channel->c, cb ? channel_poll : NULL); + utcp_set_poll_cb(channel->c, (cb || channel->aio_send) ? channel_poll : NULL); + pthread_mutex_unlock(&mesh->mutex); } void meshlink_set_channel_accept_cb(meshlink_handle_t *mesh, meshlink_channel_accept_cb_t cb) { @@ -2903,7 +3127,7 @@ void meshlink_set_channel_accept_cb(meshlink_handle_t *mesh, meshlink_channel_ac return; } - pthread_mutex_lock(&mesh->mesh_mutex); + pthread_mutex_lock(&mesh->mutex); mesh->channel_accept_cb = cb; mesh->receive_cb = channel_receive; @@ -2913,11 +3137,37 @@ void meshlink_set_channel_accept_cb(meshlink_handle_t *mesh, meshlink_channel_ac } } - pthread_mutex_unlock(&mesh->mesh_mutex); + pthread_mutex_unlock(&mesh->mutex); +} + +void meshlink_set_channel_sndbuf(meshlink_handle_t *mesh, meshlink_channel_t *channel, size_t size) { + (void)mesh; + + if(!channel) { + meshlink_errno = MESHLINK_EINVAL; + return; + } + + pthread_mutex_lock(&mesh->mutex); + utcp_set_sndbuf(channel->c, size); + pthread_mutex_unlock(&mesh->mutex); +} + +void meshlink_set_channel_rcvbuf(meshlink_handle_t *mesh, meshlink_channel_t *channel, size_t size) { + (void)mesh; + + if(!channel) { + meshlink_errno = MESHLINK_EINVAL; + return; + } + + pthread_mutex_lock(&mesh->mutex); + utcp_set_rcvbuf(channel->c, size); + pthread_mutex_unlock(&mesh->mutex); } meshlink_channel_t *meshlink_channel_open_ex(meshlink_handle_t *mesh, meshlink_node_t *node, uint16_t port, meshlink_channel_receive_cb_t cb, const void *data, size_t len, uint32_t flags) { - if(data || len) { + if(data && len) { abort(); // TODO: handle non-NULL data } @@ -2926,6 +3176,8 @@ meshlink_channel_t *meshlink_channel_open_ex(meshlink_handle_t *mesh, meshlink_n return NULL; } + pthread_mutex_lock(&mesh->mutex); + node_t *n = (node_t *)node; if(!n->utcp) { @@ -2934,20 +3186,30 @@ meshlink_channel_t *meshlink_channel_open_ex(meshlink_handle_t *mesh, meshlink_n if(!n->utcp) { meshlink_errno = errno == ENOMEM ? MESHLINK_ENOMEM : MESHLINK_EINTERNAL; + pthread_mutex_unlock(&mesh->mutex); return NULL; } } if(n->status.blacklisted) { logger(mesh, MESHLINK_ERROR, "Cannot open a channel with blacklisted node\n"); + meshlink_errno = MESHLINK_EBLACKLISTED; + pthread_mutex_unlock(&mesh->mutex); return NULL; } meshlink_channel_t *channel = xzalloc(sizeof(*channel)); channel->node = n; channel->receive_cb = cb; + + if(data && !len) { + channel->priv = (void *)data; + } + channel->c = utcp_connect_ex(n->utcp, port, channel_recv, channel, flags); + pthread_mutex_unlock(&mesh->mutex); + if(!channel->c) { meshlink_errno = errno == ENOMEM ? MESHLINK_ENOMEM : MESHLINK_EINTERNAL; free(channel); @@ -2967,7 +3229,9 @@ void meshlink_channel_shutdown(meshlink_handle_t *mesh, meshlink_channel_t *chan return; } + pthread_mutex_lock(&mesh->mutex); utcp_shutdown(channel->c, direction); + pthread_mutex_unlock(&mesh->mutex); } void meshlink_channel_close(meshlink_handle_t *mesh, meshlink_channel_t *channel) { @@ -2976,7 +3240,25 @@ void meshlink_channel_close(meshlink_handle_t *mesh, meshlink_channel_t *channel return; } + pthread_mutex_lock(&mesh->mutex); + utcp_close(channel->c); + + /* Clean up any outstanding AIO buffers. */ + for(meshlink_aio_buffer_t *aio = channel->aio_send, *next; aio; aio = next) { + next = aio->next; + aio_signal(mesh, channel, aio); + free(aio); + } + + for(meshlink_aio_buffer_t *aio = channel->aio_receive, *next; aio; aio = next) { + next = aio->next; + aio_signal(mesh, channel, aio); + free(aio); + } + + pthread_mutex_unlock(&mesh->mutex); + free(channel); } @@ -3000,9 +3282,18 @@ ssize_t meshlink_channel_send(meshlink_handle_t *mesh, meshlink_channel_t *chann // Then, preferably only if there is room in the receiver window, // kick the meshlink thread to go send packets. - pthread_mutex_lock(&mesh->mesh_mutex); - ssize_t retval = utcp_send(channel->c, data, len); - pthread_mutex_unlock(&mesh->mesh_mutex); + ssize_t retval; + + pthread_mutex_lock(&mesh->mutex); + + /* Disallow direct calls to utcp_send() while we still have AIO active. */ + if(channel->aio_send) { + retval = 0; + } else { + retval = utcp_send(channel->c, data, len); + } + + pthread_mutex_unlock(&mesh->mutex); if(retval < 0) { meshlink_errno = MESHLINK_ENETWORK; @@ -3011,6 +3302,146 @@ ssize_t meshlink_channel_send(meshlink_handle_t *mesh, meshlink_channel_t *chann return retval; } +bool meshlink_channel_aio_send(meshlink_handle_t *mesh, meshlink_channel_t *channel, const void *data, size_t len, meshlink_aio_cb_t cb, void *priv) { + if(!mesh || !channel) { + meshlink_errno = MESHLINK_EINVAL; + return false; + } + + if(!len || !data) { + meshlink_errno = MESHLINK_EINVAL; + return false; + } + + meshlink_aio_buffer_t *aio = xzalloc(sizeof(*aio)); + aio->data = data; + aio->len = len; + aio->cb.buffer = cb; + aio->priv = priv; + + pthread_mutex_lock(&mesh->mutex); + + /* Append the AIO buffer descriptor to the end of the chain */ + meshlink_aio_buffer_t **p = &channel->aio_send; + + while(*p) { + p = &(*p)->next; + } + + *p = aio; + + /* Ensure the poll callback is set, and call it right now to push data if possible */ + utcp_set_poll_cb(channel->c, channel_poll); + channel_poll(channel->c, len); + + pthread_mutex_unlock(&mesh->mutex); + + return true; +} + +bool meshlink_channel_aio_fd_send(meshlink_handle_t *mesh, meshlink_channel_t *channel, int fd, size_t len, meshlink_aio_fd_cb_t cb, void *priv) { + if(!mesh || !channel) { + meshlink_errno = MESHLINK_EINVAL; + return false; + } + + if(!len || fd == -1) { + meshlink_errno = MESHLINK_EINVAL; + return false; + } + + meshlink_aio_buffer_t *aio = xzalloc(sizeof(*aio)); + aio->fd = fd; + aio->len = len; + aio->cb.fd = cb; + aio->priv = priv; + + pthread_mutex_lock(&mesh->mutex); + + /* Append the AIO buffer descriptor to the end of the chain */ + meshlink_aio_buffer_t **p = &channel->aio_send; + + while(*p) { + p = &(*p)->next; + } + + *p = aio; + + /* Ensure the poll callback is set, and call it right now to push data if possible */ + utcp_set_poll_cb(channel->c, channel_poll); + channel_poll(channel->c, len); + + pthread_mutex_unlock(&mesh->mutex); + + return true; +} + +bool meshlink_channel_aio_receive(meshlink_handle_t *mesh, meshlink_channel_t *channel, const void *data, size_t len, meshlink_aio_cb_t cb, void *priv) { + if(!mesh || !channel) { + meshlink_errno = MESHLINK_EINVAL; + return false; + } + + if(!len || !data) { + meshlink_errno = MESHLINK_EINVAL; + return false; + } + + meshlink_aio_buffer_t *aio = xzalloc(sizeof(*aio)); + aio->data = data; + aio->len = len; + aio->cb.buffer = cb; + aio->priv = priv; + + pthread_mutex_lock(&mesh->mutex); + + /* Append the AIO buffer descriptor to the end of the chain */ + meshlink_aio_buffer_t **p = &channel->aio_receive; + + while(*p) { + p = &(*p)->next; + } + + *p = aio; + + pthread_mutex_unlock(&mesh->mutex); + + return true; +} + +bool meshlink_channel_aio_fd_receive(meshlink_handle_t *mesh, meshlink_channel_t *channel, int fd, size_t len, meshlink_aio_fd_cb_t cb, void *priv) { + if(!mesh || !channel) { + meshlink_errno = MESHLINK_EINVAL; + return false; + } + + if(!len || fd == -1) { + meshlink_errno = MESHLINK_EINVAL; + return false; + } + + meshlink_aio_buffer_t *aio = xzalloc(sizeof(*aio)); + aio->fd = fd; + aio->len = len; + aio->cb.fd = cb; + aio->priv = priv; + + pthread_mutex_lock(&mesh->mutex); + + /* Append the AIO buffer descriptor to the end of the chain */ + meshlink_aio_buffer_t **p = &channel->aio_receive; + + while(*p) { + p = &(*p)->next; + } + + *p = aio; + + pthread_mutex_unlock(&mesh->mutex); + + return true; +} + uint32_t meshlink_channel_get_flags(meshlink_handle_t *mesh, meshlink_channel_t *channel) { if(!mesh || !channel) { meshlink_errno = MESHLINK_EINVAL; @@ -3038,6 +3469,25 @@ size_t meshlink_channel_get_recvq(meshlink_handle_t *mesh, meshlink_channel_t *c return utcp_get_recvq(channel->c); } +void meshlink_set_node_channel_timeout(meshlink_handle_t *mesh, meshlink_node_t *node, int timeout) { + if(!mesh || !node) { + meshlink_errno = MESHLINK_EINVAL; + return; + } + + node_t *n = (node_t *)node; + + pthread_mutex_lock(&mesh->mutex); + + if(!n->utcp) { + n->utcp = utcp_init(channel_accept, channel_pre_accept, channel_send, n); + } + + utcp_set_user_timeout(n->utcp, timeout); + + pthread_mutex_unlock(&mesh->mutex); +} + void update_node_status(meshlink_handle_t *mesh, node_t *n) { if(n->status.reachable && mesh->channel_accept_cb && !n->utcp) { n->utcp = utcp_init(channel_accept, channel_pre_accept, channel_send, n); @@ -3046,6 +3496,16 @@ void update_node_status(meshlink_handle_t *mesh, node_t *n) { if(mesh->node_status_cb) { mesh->node_status_cb(mesh, (meshlink_node_t *)n, n->status.reachable && !n->status.blacklisted); } + + if(mesh->node_pmtu_cb) { + mesh->node_pmtu_cb(mesh, (meshlink_node_t *)n, n->minmtu); + } +} + +void update_node_pmtu(meshlink_handle_t *mesh, node_t *n) { + if(mesh->node_pmtu_cb && !n->status.blacklisted) { + mesh->node_pmtu_cb(mesh, (meshlink_node_t *)n, n->minmtu); + } } void handle_duplicate_node(meshlink_handle_t *mesh, node_t *n) { @@ -3065,7 +3525,7 @@ void meshlink_enable_discovery(meshlink_handle_t *mesh, bool enable) { return; } - pthread_mutex_lock(&mesh->mesh_mutex); + pthread_mutex_lock(&mesh->mutex); if(mesh->discovery == enable) { goto end; @@ -3082,7 +3542,7 @@ void meshlink_enable_discovery(meshlink_handle_t *mesh, bool enable) { mesh->discovery = enable; end: - pthread_mutex_unlock(&mesh->mesh_mutex); + pthread_mutex_unlock(&mesh->mutex); #else (void)mesh; (void)enable; @@ -3090,31 +3550,53 @@ end: #endif } +void meshlink_set_dev_class_timeouts(meshlink_handle_t *mesh, dev_class_t devclass, int pinginterval, int pingtimeout) { + if(!mesh || devclass < 0 || devclass >= DEV_CLASS_COUNT) { + meshlink_errno = EINVAL; + return; + } + + if(pinginterval < 1 || pingtimeout < 1 || pingtimeout > pinginterval) { + meshlink_errno = EINVAL; + return; + } + + pthread_mutex_lock(&mesh->mutex); + mesh->dev_class_traits[devclass].pinginterval = pinginterval; + mesh->dev_class_traits[devclass].pingtimeout = pingtimeout; + pthread_mutex_unlock(&mesh->mutex); +} + void handle_network_change(meshlink_handle_t *mesh, bool online) { (void)online; - if(!mesh->connections) { + if(!mesh->connections || !mesh->loop.running) { return; } retry(mesh); } +void call_error_cb(meshlink_handle_t *mesh, meshlink_errno_t meshlink_errno) { + // We should only call the callback function if we are in the background thread. + if(!mesh->error_cb) { + return; + } + + if(!mesh->threadstarted) { + return; + } + + if(mesh->thread == pthread_self()) { + mesh->error_cb(mesh, meshlink_errno); + } +} + + static void __attribute__((constructor)) meshlink_init(void) { crypto_init(); - unsigned int seed; - randomize(&seed, sizeof(seed)); - srand(seed); } static void __attribute__((destructor)) meshlink_exit(void) { crypto_exit(); } - -/// Device class traits -dev_class_traits_t dev_class_traits[_DEV_CLASS_MAX + 1] = { - { .min_connects = 3, .max_connects = 10000, .edge_weight = 1 }, // DEV_CLASS_BACKBONE - { .min_connects = 3, .max_connects = 100, .edge_weight = 3 }, // DEV_CLASS_STATIONARY - { .min_connects = 3, .max_connects = 3, .edge_weight = 6 }, // DEV_CLASS_PORTABLE - { .min_connects = 1, .max_connects = 1, .edge_weight = 9 }, // DEV_CLASS_UNKNOWN -};