X-Git-Url: http://git.meshlink.io/?p=meshlink;a=blobdiff_plain;f=src%2Fmeshlink.c;h=7a01bd0ffae8f921f7e3b28457e26f6198b7b6b2;hp=134abd83da5e548c3640a1dc056ec107ea189063;hb=477973fd217114dec8218cb7f8d99655272a3669;hpb=53522b1c11222273c7b41f72b374e759d13b2165 diff --git a/src/meshlink.c b/src/meshlink.c index 134abd83..7a01bd0f 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); @@ -531,6 +531,10 @@ int check_port(meshlink_handle_t *mesh) { } static bool write_main_config_files(meshlink_handle_t *mesh) { + if(!mesh->confbase) { + return true; + } + uint8_t buf[4096]; /* Write the main config file */ @@ -548,7 +552,7 @@ static bool write_main_config_files(meshlink_handle_t *mesh) { config_t config = {buf, packmsg_output_size(&out, buf)}; - if(!main_config_write(mesh, &config)) { + if(!main_config_write(mesh, "current", &config, mesh->config_key)) { return false; } @@ -571,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) { @@ -595,10 +599,10 @@ 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)) { + if(!config_init(mesh, "current")) { return false; } @@ -652,11 +656,16 @@ static bool finalize_join(meshlink_handle_t *mesh, const void *buf, uint16_t len node_add(mesh, n); - if(!config_write(mesh, n->name, &config)) { + if(!config_write(mesh, "current", n->name, &config, mesh->config_key)) { return false; } } + /* 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); @@ -740,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; @@ -836,6 +846,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 @@ -856,8 +867,8 @@ static void add_local_addresses(meshlink_handle_t *mesh) { } static bool meshlink_setup(meshlink_handle_t *mesh) { - if(!config_init(mesh)) { - logger(mesh, MESHLINK_ERROR, "Could not set up configuration in %s: %s\n", mesh->confbase, strerror(errno)); + if(!config_init(mesh, "current")) { + logger(mesh, MESHLINK_ERROR, "Could not set up configuration in %s/current: %s\n", mesh->confbase, strerror(errno)); meshlink_errno = MESHLINK_ESTORAGE; return false; } @@ -880,6 +891,13 @@ static bool meshlink_setup(meshlink_handle_t *mesh) { mesh->self->ecdsa = ecdsa_set_public_key(ecdsa_get_public_key(mesh->private_key)); 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)); + meshlink_errno = MESHLINK_ESTORAGE; + return false; + } + + /* Ensure the configuration directory metadata is on disk */ + if(!config_sync(mesh, "current")) { return false; } @@ -902,7 +920,7 @@ static bool meshlink_read_config(meshlink_handle_t *mesh) { config_t config; - if(!main_config_read(mesh, &config)) { + if(!main_config_read(mesh, "current", &config, mesh->config_key)) { logger(NULL, MESHLINK_ERROR, "Could not read main configuration file!"); return false; } @@ -996,16 +1014,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; @@ -1051,6 +1070,85 @@ bool meshlink_open_params_set_storage_key(meshlink_open_params_t *params, const return true; } +bool meshlink_encrypted_key_rotate(meshlink_handle_t *mesh, const void *new_key, size_t new_keylen) { + if(!mesh || !new_key || !new_keylen) { + logger(mesh, MESHLINK_ERROR, "Invalid arguments given!\n"); + meshlink_errno = MESHLINK_EINVAL; + return false; + } + + pthread_mutex_lock(&(mesh->mesh_mutex)); + + // Create hash for the new key + void *new_config_key; + new_config_key = xmalloc(CHACHA_POLY1305_KEYLEN); + + 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)); + return false; + } + + // Copy contents of the "current" confbase sub-directory to "new" confbase sub-directory with the 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)); + return false; + } + + devtool_keyrotate_probe(1); + + main_config_unlock(mesh); + + // Rename confbase/current/ to confbase/old + + if(!config_rename(mesh, "current", "old")) { + 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)); + return false; + } + + devtool_keyrotate_probe(2); + + // Rename confbase/new/ to confbase/current + + if(!config_rename(mesh, "new", "current")) { + 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)); + return false; + } + + devtool_keyrotate_probe(3); + + if(!main_config_lock(mesh)) { + pthread_mutex_unlock(&(mesh->mesh_mutex)); + return false; + } + + // Cleanup the "old" confbase sub-directory + + if(!config_destroy(mesh->confbase, "old")) { + pthread_mutex_unlock(&(mesh->mesh_mutex)); + return false; + } + + // Change the mesh handle key with new key + + free(mesh->config_key); + mesh->config_key = new_config_key; + + pthread_mutex_unlock(&(mesh->mesh_mutex)); + + return true; +} + void meshlink_open_params_free(meshlink_open_params_t *params) { if(!params) { meshlink_errno = MESHLINK_EINVAL; @@ -1064,6 +1162,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"); @@ -1092,7 +1198,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; @@ -1109,7 +1216,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; @@ -1151,7 +1259,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; @@ -1175,6 +1283,12 @@ 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)); + + memcpy(mesh->dev_class_traits, default_class_traits, sizeof(default_class_traits)); if(usingname) { mesh->name = xstrdup(params->name); @@ -1206,7 +1320,7 @@ meshlink_handle_t *meshlink_open_ex(const meshlink_open_params_t *params) { // If no configuration exists yet, create it. - if(!main_config_exists(mesh)) { + if(!meshlink_confbase_exists(mesh)) { if(!meshlink_setup(mesh)) { logger(NULL, MESHLINK_ERROR, "Cannot create initial configuration\n"); meshlink_close(mesh); @@ -1296,21 +1410,42 @@ 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 } +#if HAVE_CATTA + + if(mesh->discovery) { + discovery_start(mesh); + } + +#endif + pthread_mutex_lock(&(mesh->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->mesh_mutex)); + +#if HAVE_CATTA + + // Stop discovery + if(mesh->discovery) { + discovery_stop(mesh); + } + +#endif + return NULL; } @@ -1327,6 +1462,9 @@ bool meshlink_start(meshlink_handle_t *mesh) { pthread_mutex_lock(&(mesh->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)); @@ -1366,20 +1504,9 @@ bool meshlink_start(meshlink_handle_t *mesh) { return false; } + pthread_cond_wait(&mesh->cond, &mesh->mesh_mutex); mesh->threadstarted = true; -#if HAVE_CATTA - - if(mesh->discovery) { - discovery_start(mesh); - } - -#endif - - 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)); return true; } @@ -1393,15 +1520,6 @@ void meshlink_stop(meshlink_handle_t *mesh) { pthread_mutex_lock(&(mesh->mesh_mutex)); logger(mesh, MESHLINK_DEBUG, "meshlink_stop called\n"); -#if HAVE_CATTA - - // Stop discovery - if(mesh->discovery) { - discovery_stop(mesh); - } - -#endif - // Shut down the main thread event_loop_stop(&mesh->loop); @@ -1488,15 +1606,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->mesh_mutex); + pthread_mutex_destroy(&mesh->mesh_mutex); + memset(mesh, 0, sizeof(*mesh)); free(mesh); @@ -1508,7 +1634,21 @@ bool meshlink_destroy(const char *confbase) { return false; } - return config_destroy(confbase); + if(!config_destroy(confbase, "current")) { + logger(NULL, MESHLINK_ERROR, "Cannot remove confbase sub-directories %s: %s\n", confbase, strerror(errno)); + return false; + } + + config_destroy(confbase, "new"); + config_destroy(confbase, "old"); + + if(rmdir(confbase) && errno != ENOENT) { + logger(NULL, MESHLINK_ERROR, "Cannot remove directory %s: %s\n", confbase, strerror(errno)); + meshlink_errno = MESHLINK_ESTORAGE; + return false; + } + + return true; } void meshlink_set_receive_cb(meshlink_handle_t *mesh, meshlink_receive_cb_t cb) { @@ -1544,6 +1684,17 @@ void meshlink_set_node_status_cb(meshlink_handle_t *mesh, meshlink_node_status_c pthread_mutex_unlock(&(mesh->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->mesh_mutex)); + mesh->node_pmtu_cb = cb; + pthread_mutex_unlock(&(mesh->mesh_mutex)); +} + void meshlink_set_node_duplicate_cb(meshlink_handle_t *mesh, meshlink_node_duplicate_cb_t cb) { if(!mesh) { meshlink_errno = MESHLINK_EINVAL; @@ -1567,6 +1718,17 @@ void meshlink_set_log_cb(meshlink_handle_t *mesh, meshlink_log_level_t level, me } } +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->mesh_mutex)); + mesh->error_cb = cb; + pthread_mutex_unlock(&(mesh->mesh_mutex)); +} + bool meshlink_send(meshlink_handle_t *mesh, meshlink_node_t *destination, const void *data, size_t len) { meshlink_packethdr_t *hdr; @@ -1626,8 +1788,9 @@ bool meshlink_send(meshlink_handle_t *mesh, meshlink_node_t *destination, const 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) { @@ -1637,6 +1800,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) { @@ -1708,6 +1873,11 @@ meshlink_node_t *meshlink_get_node(meshlink_handle_t *mesh, const char *name) { pthread_mutex_lock(&(mesh->mesh_mutex)); node = (meshlink_node_t *)lookup_node(mesh, (char *)name); // TODO: make lookup_node() use const pthread_mutex_unlock(&(mesh->mesh_mutex)); + + if(!node) { + meshlink_errno = MESHLINK_ENOENT; + } + return node; } @@ -1722,6 +1892,11 @@ meshlink_submesh_t *meshlink_get_submesh(meshlink_handle_t *mesh, const char *na pthread_mutex_lock(&(mesh->mesh_mutex)); submesh = (meshlink_submesh_t *)lookup_submesh(mesh, name); pthread_mutex_unlock(&(mesh->mesh_mutex)); + + if(!submesh) { + meshlink_errno = MESHLINK_ENOENT; + } + return submesh; } @@ -1799,7 +1974,7 @@ static meshlink_node_t **meshlink_get_all_nodes_by_condition(meshlink_handle_t * static bool search_node_by_dev_class(const node_t *node, const void *condition) { dev_class_t *devclass = (dev_class_t *)condition; - if(*devclass == node->devclass) { + if(*devclass == (dev_class_t)node->devclass) { return true; } @@ -1815,7 +1990,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; } @@ -2001,7 +2176,13 @@ int meshlink_get_port(meshlink_handle_t *mesh) { return -1; } - return atoi(mesh->myport); + int port; + + pthread_mutex_lock(&(mesh->mesh_mutex)); + port = atoi(mesh->myport); + pthread_mutex_unlock(&(mesh->mesh_mutex)); + + return port; } bool meshlink_set_port(meshlink_handle_t *mesh, int port) { @@ -2033,9 +2214,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); @@ -2043,18 +2221,28 @@ 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; + xasprintf(&mesh->myport, "%d", port); if(!node_read_public_key(mesh, mesh->self)) { logger(NULL, MESHLINK_ERROR, "Could not read our host configuration file!"); 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)); @@ -2096,7 +2284,7 @@ char *meshlink_invite_ex(meshlink_handle_t *mesh, meshlink_submesh_t *submesh, c } // Ensure no host configuration file with that name exists - if(config_exists(mesh, name)) { + 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)); @@ -2163,10 +2351,11 @@ 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, mesh->self->name, &configs[count])) { + if(config_read(mesh, "current", mesh->self->name, &configs[count], mesh->config_key)) { count++; } @@ -2180,7 +2369,7 @@ char *meshlink_invite_ex(meshlink_handle_t *mesh, meshlink_submesh_t *submesh, c config_t config = {outbuf, packmsg_output_size(&inv, outbuf)}; - if(!invitation_write(mesh, cookiehash, &config)) { + 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)); @@ -2210,7 +2399,15 @@ bool meshlink_join(meshlink_handle_t *mesh, const char *invitation) { //Before doing meshlink_join make sure we are not connected to another mesh if(mesh->threadstarted) { - logger(mesh, MESHLINK_DEBUG, "Already connected to a mesh\n"); + logger(mesh, MESHLINK_ERROR, "Cannot join while started\n"); + meshlink_errno = MESHLINK_EINVAL; + pthread_mutex_unlock(&(mesh->mesh_mutex)); + return false; + } + + // Refuse to join a mesh if we are already part of one. We are part of one if we know at least one other node. + 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)); return false; @@ -2574,18 +2771,24 @@ bool meshlink_import(meshlink_handle_t *mesh, const char *data) { break; } - config_write(mesh, n->name, &config); + config_write(mesh, "current", n->name, &config, mesh->config_key); node_add(mesh, n); } pthread_mutex_unlock(&(mesh->mesh_mutex)); + free(buf); + if(!packmsg_done(&in)) { logger(mesh, MESHLINK_ERROR, "Invalid data\n"); meshlink_errno = MESHLINK_EPEER; return false; } + if(!config_sync(mesh, "current")) { + return false; + } + return true; } @@ -2615,6 +2818,8 @@ void meshlink_blacklist(meshlink_handle_t *mesh, meshlink_node_t *node) { 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 @@ -2658,6 +2863,11 @@ void meshlink_whitelist(meshlink_handle_t *mesh, meshlink_node_t *node) { n->status.blacklisted = false; node_write_config(mesh, n); + config_sync(mesh, "current"); + + if(n->status.reachable) { + update_node_status(mesh, n); + } pthread_mutex_unlock(&(mesh->mesh_mutex)); return; @@ -2694,6 +2904,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; @@ -2706,8 +2928,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; @@ -2777,16 +3039,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(channel->poll_cb) { - channel->poll_cb(mesh, channel, len); + if(result > 0) { + sent = utcp_send(connection, buf, result); + } else { + sent = result; + } + } + + 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->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->mesh_mutex); } void meshlink_set_channel_accept_cb(meshlink_handle_t *mesh, meshlink_channel_accept_cb_t cb) { @@ -2808,6 +3123,32 @@ void meshlink_set_channel_accept_cb(meshlink_handle_t *mesh, meshlink_channel_ac pthread_mutex_unlock(&mesh->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->mesh_mutex); + utcp_set_sndbuf(channel->c, size); + pthread_mutex_unlock(&mesh->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->mesh_mutex); + utcp_set_rcvbuf(channel->c, size); + pthread_mutex_unlock(&mesh->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) { abort(); // TODO: handle non-NULL data @@ -2818,6 +3159,8 @@ meshlink_channel_t *meshlink_channel_open_ex(meshlink_handle_t *mesh, meshlink_n return NULL; } + pthread_mutex_lock(&mesh->mesh_mutex); + node_t *n = (node_t *)node; if(!n->utcp) { @@ -2826,12 +3169,14 @@ 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->mesh_mutex); return NULL; } } if(n->status.blacklisted) { logger(mesh, MESHLINK_ERROR, "Cannot open a channel with blacklisted node\n"); + pthread_mutex_unlock(&mesh->mesh_mutex); return NULL; } @@ -2840,6 +3185,8 @@ meshlink_channel_t *meshlink_channel_open_ex(meshlink_handle_t *mesh, meshlink_n channel->receive_cb = cb; channel->c = utcp_connect_ex(n->utcp, port, channel_recv, channel, flags); + pthread_mutex_unlock(&mesh->mesh_mutex); + if(!channel->c) { meshlink_errno = errno == ENOMEM ? MESHLINK_ENOMEM : MESHLINK_EINTERNAL; free(channel); @@ -2859,7 +3206,9 @@ void meshlink_channel_shutdown(meshlink_handle_t *mesh, meshlink_channel_t *chan return; } + pthread_mutex_lock(&mesh->mesh_mutex); utcp_shutdown(channel->c, direction); + pthread_mutex_unlock(&mesh->mesh_mutex); } void meshlink_channel_close(meshlink_handle_t *mesh, meshlink_channel_t *channel) { @@ -2868,7 +3217,25 @@ void meshlink_channel_close(meshlink_handle_t *mesh, meshlink_channel_t *channel return; } + pthread_mutex_lock(&mesh->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->mesh_mutex); + free(channel); } @@ -2892,8 +3259,17 @@ 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. + ssize_t retval; + pthread_mutex_lock(&mesh->mesh_mutex); - ssize_t retval = utcp_send(channel->c, data, len); + + /* 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->mesh_mutex); if(retval < 0) { @@ -2903,6 +3279,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->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->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->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->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->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->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->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->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; @@ -2938,6 +3454,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) { @@ -2982,21 +3508,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->mesh_mutex); + mesh->dev_class_traits[devclass].pinginterval = pinginterval; + mesh->dev_class_traits[devclass].pingtimeout = pingtimeout; + pthread_mutex_unlock(&mesh->mesh_mutex); +} + +void handle_network_change(meshlink_handle_t *mesh, bool online) { + (void)online; + + 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 -};