c-libp2p/routing/kademlia.c

/* For crypt */
#define _GNU_SOURCE

#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <string.h>
#include <unistd.h>
#include <fcntl.h>
#include <sys/time.h>
#include <arpa/inet.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netdb.h>
#include <sys/signal.h>
#include <pthread.h>
#include <libp2p/crypto/sha256.h>
#include <libp2p/routing/kademlia.h>
#include <libp2p/routing/dht.h>
#include <multiaddr/multiaddr.h>

extern FILE *dht_debug;

#define MAX_BOOTSTRAP_NODES 20
static struct sockaddr_storage bootstrap_nodes[MAX_BOOTSTRAP_NODES];
static int num_bootstrap_nodes = 0;

struct bs_struct {
    char *ip;
    uint16_t port;
} bootstrap_list[] = {
};

pthread_t pth_kademlia, pth_announce;
time_t tosleep = 0;
int kfd = -1;
int net_family = 0;
volatile int8_t searching = 0; // search lock, -1 to busy, 0 to free, 1 to running.
volatile char hash[20];     // hash to be search or announce.
volatile uint16_t announce_port = 0;
volatile int8_t closing = 0;

#define ANNOUNCE_WAIT_TIME		(28 * 60) // Wait 28 minutes.
#define ANNOUNCE_WAIT_TOLERANCE		60
struct announce_struct {
    unsigned char hash[sizeof hash];
    uint16_t port;
    unsigned int time;
    struct announce_struct *next;
} *announce_list = NULL;

#define DHT_MAX_IPV4	50
#define DHT_MAX_IPV6	10

struct ipv4_struct {
    uint32_t ip;
    uint16_t port;
};

struct ipv6_struct {
    uint8_t  ip[16];
    uint16_t port;
};

struct search_struct {
    unsigned char hash[sizeof hash];
    uint8_t ipv4_count;
    uint8_t ipv6_count;
    struct ipv4_struct ipv4[DHT_MAX_IPV4];
    struct ipv6_struct ipv6[DHT_MAX_IPV6];
    struct search_struct *next;
} *search_result = NULL;

/***
 * The call-back function is called by the DHT whenever something
 * interesting happens.  Right now, it only happens when we get a new value or
 * when a search completes, but this may be extended in future versions.
 * @param closure
 * @param event the event
 * @param info_hash the hash to work with
 * @param data the new value to be added
 * @param data_len the length of the data
 */
static void callback(void *closure, int event, const unsigned char *info_hash, const void *data, size_t data_len) {
    struct search_struct *sp, *rp = NULL; // struct pointer and result pointer

    switch (event) {
        case DHT_EVENT_VALUES:
        case DHT_EVENT_VALUES6:
            if (dht_debug) {
                fprintf(dht_debug, "Received %d values.\n", (int)(data_len / 6));
            }
            // Find the item in the list.
            for (rp = search_result ; rp ; rp = rp->next) {
                if (memcmp(rp->hash, info_hash, sizeof hash) == 0) { // Found.
                    int i;
                    if (event == DHT_EVENT_VALUES) { // IPv4
                        struct ipv4_struct ipv4;
                        if (rp->ipv4_count == DHT_MAX_IPV4) { // Full
                            return;
                        }
                        // Make sure the data is in struct format.
                        memcpy(&ipv4.ip, data, 4);
                        memcpy(&ipv4.port, data+4, 2);
                        ipv4.port = ntohs(ipv4.port);
                        for (i = 0 ; i < rp->ipv4_count ; i++) {
                            if (memcmp(&rp->ipv4[i], &ipv4, sizeof ipv4) == 0) {
                                return; // Alread in the list.
                            }
                        }
                        // Not in the list, then add.
                        memcpy(&rp->ipv4[rp->ipv4_count], &ipv4, sizeof ipv4);
                        rp->ipv4_count++;
                    } else { // IPv6
                        struct ipv6_struct ipv6;
                        if (rp->ipv6_count == DHT_MAX_IPV6) { // Full
                            return;
                        }
                        // Make sure the data is in struct format.
                        memcpy(&ipv6.ip, data, 16);
                        memcpy(&ipv6.port, data+16, 2);
                        for (i = 0 ; i < rp->ipv6_count ; i++) {
                            if (memcmp(&rp->ipv6[i], &ipv6, sizeof ipv6) == 0) {
                                return; // Alread in the list.
                            }
                        }
                        // Not in the list, then add.
                        memcpy(&rp->ipv6[rp->ipv6_count], &ipv6, sizeof ipv6);
                        rp->ipv6_count++;
                    }
                    return;
                }
            }
            break; // Not found, how can EVENT_VALUE may occur before SEARCH_DONE?
        case DHT_EVENT_SEARCH_DONE:
        case DHT_EVENT_SEARCH_DONE6:
            if (dht_debug) {
                fprintf(dht_debug, "Search done.\n");
            }
            if (search_result) {
                // Try to find the item in the list.
                for (sp = search_result ; sp->next ; sp = sp->next) {
                    if (memcmp(sp->hash, info_hash, sizeof hash) == 0) { // Found.
                        rp = sp;
                        break;
                    }
                }
                if (!rp) {
                    rp = malloc(sizeof(struct search_struct));
                    if (!rp) return; // Abort, out of memory.
                    memset(rp, 0, sizeof(struct search_struct));
                    memcpy(rp->hash, info_hash, sizeof hash);
                    sp->next = rp; // Insert in the list.
                }
            } else {
                rp = malloc(sizeof(struct search_struct));
                if (!rp) return; // Abort, out of memory.
                memset(rp, 0, sizeof(struct search_struct));
                memcpy(rp->hash, info_hash, sizeof hash);
                search_result = rp; // Insert first item in the list.
            }
            break;
        default:
            break;
    }
}

int start_kademlia_multiaddress(struct MultiAddress* address, char* peer_id, int timeout, struct Libp2pVector* bootstrap_addresses) {
	int port = multiaddress_get_ip_port(address);
	int family = multiaddress_get_ip_family(address);
	int fd = socket(family, SOCK_DGRAM, 0);
	if (fd < 0)
		return 0;
	struct sockaddr_in serv_addr;
	serv_addr.sin_family = family;
	serv_addr.sin_addr.s_addr = htonl(INADDR_ANY);
	serv_addr.sin_port = port;
	if (bind(fd, (struct sockaddr*)&serv_addr, sizeof(serv_addr)) != 0)
		return 0;
	return start_kademlia(fd, family, peer_id, timeout, bootstrap_addresses);
}

/***
 * Start the kademlia service
 * @param net_fd the file descriptor of the address/socket already bound
 * @param family ip4 or ip6
 * @param peer_id the first 20 chars of the public PeerID in a null terminated string
 * @param timeout seconds before a select() timeout
 */
int start_kademlia(int net_fd, int family, char* peer_id, int timeout, struct Libp2pVector* bootstrap_addresses)
{
    int rc, i, len;
    unsigned char id[sizeof hash];
    struct sockaddr_in sa;

    dht_debug = stderr;

    len = bootstrap_addresses->total;

    if (len > MAX_BOOTSTRAP_NODES) {
        len = MAX_BOOTSTRAP_NODES; // limit array length
    }

    memset(&sa, 0, sizeof sa);
    char* ip = NULL;
    for (i = 0 ; i < len ; i++) {
    	struct MultiAddress* addr = (struct MultiAddress*)libp2p_utils_vector_get(bootstrap_addresses, i);
    	if (ip != NULL)
    		free(ip);
    	if (multiaddress_is_ip(addr)) {
    		multiaddress_get_ip_address(addr, &ip);
    		if (family == AF_INET6 && multiaddress_is_ip6(addr)) {
    			if (inet_pton(AF_INET6, ip, &(sa.sin_addr.s_addr)) == 1)
    				sa.sin_family = AF_INET6;
    			else
    				continue;
    		} else {
    			if (inet_pton(AF_INET , ip, &(sa.sin_addr.s_addr)) == 1)
					sa.sin_family = AF_INET;
    			else
    				continue;
    		}
    	} else {
            continue; // not an ipv6 or ipv4?
        }

        sa.sin_port = htons (multiaddress_get_ip_port(addr));

        memcpy(&bootstrap_nodes[num_bootstrap_nodes++], &sa, sizeof(sa));
    }
    if (ip != NULL)
    	free(ip);

    dht_hash (id, sizeof(id), peer_id, strlen(peer_id), NULL, 0, NULL, 0);

    if (family == AF_INET6) {
        rc = dht_init(-1, net_fd, id, NULL);
    } else {
        rc = dht_init(net_fd, -1, id, NULL);
    }
    if (rc < 0) {
        return rc;
    }

    /* For bootstrapping, we need an initial list of nodes.  This could be
       hard-wired, but can also be obtained from the nodes key of a torrent
       file, or from the PORT bittorrent message.

       Dht_ping_node is the brutal way of bootstrapping -- it actually
       sends a message to the peer.  If you're going to bootstrap from
       a massive number of nodes (for example because you're restoring from
       a dump) and you already know their ids, it's better to use
       dht_insert_node.  If the ids are incorrect, the DHT will recover. */
    for(i = 0; i < num_bootstrap_nodes; i++) {
    	// for debugging
    	int retVal =
        dht_ping_node((struct sockaddr*)&bootstrap_nodes[i],
                      sizeof (bootstrap_nodes[i]));
    	fprintf(stderr, "ping returned %d\n", retVal);
        usleep(random() % 100000);
    }

    // TODO: Read cache nodes from file and load using dht_insert_node.

    kfd = net_fd;
    net_family = family;
    tosleep = timeout;

    rc = pthread_create(&pth_kademlia, NULL, kademlia_thread, NULL);
    if (rc) {
        return rc; // error
    }

    return pthread_create(&pth_announce, NULL, announce_thread, NULL);
}

void stop_kademlia (void)
{
    if (kfd != -1) {
        closing = 1;

        pthread_cancel(pth_announce);

        // Wait kademlia_thread finish.
        pthread_join(pth_kademlia, NULL);

        dht_uninit();

        close (kfd);
        kfd = -1;
    }
}

void *kademlia_thread (void *ptr)
{
    int rc;
    struct timeval tv;
    fd_set readfds;
    char buf[4096];
    struct sockaddr from;
    socklen_t fromlen;

    for(;;) {
        tv.tv_sec = tosleep;
        tv.tv_usec = random() % 1000000;

        FD_ZERO(&readfds);
        FD_SET(kfd, &readfds);
        rc = select(kfd + 1, &readfds, NULL, NULL, &tv);
        if(rc < 0) {
            if(errno != EINTR) {
                perror("select");
                sleep(1);
            }
        }

        if(rc > 0 && FD_ISSET(kfd, &readfds)) {
            fromlen = sizeof(from);
            rc = recvfrom(kfd, buf, sizeof(buf) - 1, 0,
                          (struct sockaddr*)&from, &fromlen);
            if (rc < 0) {
            	if (errno == EAGAIN || errno == EWOULDBLOCK) {
            		continue;
            	} else {
            		fprintf(stderr, "kademlia_thread:recvfrom failed with %d\n", errno);
            		continue;
            	}
            }
            buf[rc] = '\0';
            rc = dht_periodic(buf, rc, (struct sockaddr*)&from, fromlen,
                              &tosleep, callback, NULL);
        } else {
            rc = dht_periodic(NULL, 0, NULL, 0, &tosleep, callback, NULL);
        }
        if(rc < 0) {
            if(errno == EINTR) {
                continue;
            } else {
                perror("dht_periodic");
                if(rc == EINVAL || rc == EFAULT)
                    abort();
                tosleep = 1;
            }
        }

        /* This is how you trigger a search for a torrent hash.  If port
           (the second argument) is non-zero, it also performs an announce.
           Since peers expire announced data after 30 minutes, it's a good
           idea to reannounce every 28 minutes or so. */
        if(searching > 0) {
            unsigned char h[sizeof hash];
            int i;
            for (i = 0 ; i < sizeof hash ; i++) {
                h[i] = hash[i]; // Copy hash array to new array so can call
                                // dht_search without volatile variable.
            }
            dht_search(h, announce_port, net_family, callback, NULL);
            searching = 0;
        }
        if(closing) {
            // TODO: Create a routine to save the cache nodes in the file sometimes and before closing.
            return 0; // end thread.
        }
    }
    return (void*)1;
}

/**
 * Search for a hash
 * @param id the hash to look for
 * @param port the port if it is available
 * @param to the time out
 * @returns the time left
 */
int search_kademlia_internal (unsigned char* id, int port, int to)
{
    int i;

    while (searching != 0) {
        i = random() % 100000;
        if (i > to) {
            return 0; // timeout waiting a chance
        }
        usleep(i);
        to -= i;
    }

    searching = -1; // lock.

    for (i = 0 ; i < sizeof hash ; i++) {
        hash[i] = id[i];
    }

    announce_port = port;

    searching = 1; // search.

    return to;
}

void *announce_thread (void *ptr)
{
    unsigned int wait;
    struct announce_struct *n, *p;

    for(;;) {
        if (announce_list) {
            unsigned int now, minus_time = ((unsigned int) -1);

            p = NULL;
            // find max wait value.
            for (n = announce_list ; n ; n = n->next) {
                if (n->time < minus_time) {
                    minus_time = n->time;
                    p = n;
                }
            }
            if (p) {
                now = time(NULL);
                if ((minus_time + ANNOUNCE_WAIT_TIME) > (now + ANNOUNCE_WAIT_TOLERANCE)) {
                    wait = ANNOUNCE_WAIT_TIME - (now - minus_time);
                    sleep (wait);
                } else {
                    if (p) {
                        search_kademlia_internal (p->hash, p->port, ANNOUNCE_WAIT_TOLERANCE * 1000000);
                        p->time = time(NULL);
                    }
                }
                continue;
            }
        }
        // Empty list, just wait.
        sleep (ANNOUNCE_WAIT_TIME);
    }
    return (void*)1;
}

int announce_kademlia (char* peer_id, uint16_t port)
{
    unsigned char id[sizeof hash];
    struct announce_struct *n, *p;

    dht_hash (id, sizeof(id), peer_id, strlen(peer_id), NULL, 0, NULL, 0);

    for (n = announce_list ; n ; n = n->next) {
        if (memcmp(n->hash, id, sizeof id) == 0) {
            return 0; // Already on the list.
        }
        if (! (n->next)) {
            break; // Keep n->next at the insertion point.
        }
    }


    if ((p = malloc (sizeof(struct announce_struct))) == NULL) {
        return 0; // Fail to alloc.
    }

    search_kademlia_internal (id, port, ANNOUNCE_WAIT_TOLERANCE * 1000000);

    memcpy(p->hash, id, sizeof id);
    p->port = port;
    p->next = NULL;
    p->time = time(NULL);

    if (!announce_list) {
        announce_list = p;
    } else {
        n->next = p;
    }

    return 1; // Announced and added to the list.
}

struct MultiAddress** search_kademlia(char* peer_id, int timeout)
{
    unsigned char id[sizeof hash];
    int i, to = timeout * 1000000;
    struct search_struct *rp; // result pointer
    struct MultiAddress **ret;

    if (kfd == -1) {
        return NULL; // start thread first.
    }

    dht_hash (id, sizeof(id), peer_id, strlen(peer_id), NULL, 0, NULL, 0);

    //TODO: Is this the right place to ask the net?
    dht_search(peer_id, 0, AF_INET, NULL, NULL);
    to = search_kademlia_internal (id, 0, to);
    if (to == 0) return NULL; // time out.

    // Wait for search completion.
    for(;;) {
        i = random() % 100000;
        if (i > to) {
            return NULL; // timeout.
        }
        usleep(i);
        for (rp = search_result ; rp ; rp = rp->next) {
            if (memcmp(rp->hash, id, sizeof hash) == 0) { // Found.
                char ipstr[INET6_ADDRSTRLEN + 1];
                char str[sizeof ipstr + 16];
                int c = 0;

                // Wait for result or time out.
                while (to > 0 &&
                       search_result->ipv4_count == 0 &&
                       search_result->ipv6_count == 0) {
                         to = search_kademlia_internal (id, 0, to); // Repeat search to collect result.
                         usleep(2000000); // Wait a few seconds for the result.
                         to -= 2000000;
                }

                if (search_result->ipv4_count == 0 ||
                    search_result->ipv6_count == 0) return NULL; // no result.

                ret = calloc(search_result->ipv4_count + search_result->ipv6_count + 1, // IPv4 + IPv6 itens and a NULL terminator.
                             sizeof (struct MultiAddress*)); // array of pointer.
                if (!ret) {
                    return NULL;
                }

                for (i = 0 ; i < search_result->ipv4_count ; i++) {
                    if (inet_ntop(AF_INET, &search_result->ipv4[i].ip, ipstr, sizeof ipstr)) {
                        snprintf (str, sizeof str, "/ip4/%s/tcp/%d", ipstr, search_result->ipv4[i].port);
                        if (dht_debug) {
                            fprintf(dht_debug, "SEARCH %s (%d) = %s\n", peer_id, c, str);
                        }
                        ret[c] = multiaddress_new_from_string (str);
                        if (ret[c] > 0) { // Sucess.
                            c++;
                        }
                    }
                }
                for (i = 0 ; i < search_result->ipv6_count ; i++) {
                    if (inet_ntop(AF_INET6, search_result->ipv6[i].ip, ipstr, sizeof ipstr)) {
                        snprintf (str, sizeof str, "/ip6/%s/tcp/%d", ipstr, search_result->ipv6[i].port);
                        if (dht_debug) {
                            fprintf(dht_debug, "SEARCH %s (%d) = %s\n", peer_id, c, str);
                        }
                        ret[c] = multiaddress_new_from_string (str);
                        if (ret[c] > 0) { // Sucess.
                            c++;
                        }
                    }
                }
                ret[c] = NULL; // NULL terminator.
                return ret;
            }
        }
        to -= i;
    }

    return NULL;
}

int ping_kademlia (char *ip, uint16_t port)
{
    struct sockaddr_in sa;

    if (inet_pton(AF_INET6, ip, &(sa.sin_addr.s_addr)) == 1) {
        sa.sin_family = AF_INET6;
    } if (inet_pton(AF_INET, ip, &(sa.sin_addr.s_addr)) == 1) {
        sa.sin_family = AF_INET;
    } else {
        return 0;
    }

    sa.sin_port = htons (port);

    dht_ping_node((struct sockaddr*)&sa, sizeof sa);
    //usleep(random() % 100000);

    return 1;
}

/* Functions called by the DHT. */

int dht_blacklisted (const struct sockaddr *sa, int salen)
{
    return 0;
}

/* We need to provide a reasonably strong cryptographic hashing function.
   libp2p_crypto_hashing_sha256 */
void dht_hash (void *hash_return, int hash_size,
               const void *v1, int len1,
               const void *v2, int len2,
               const void *v3, int len3)
{
    int len = len1 + len2 + len3;
    unsigned char *in, out[32];

    if (!hash_return || hash_size==0 || len==0) {
        return; // invalid param.
    }

    in = malloc (len);

    if (in) {
        memcpy(in, v1, len1);
        memcpy(in+len1, v2, len2);
        memcpy(in+len1+len2, v3, len3);
        if ( libp2p_crypto_hashing_sha256 (in, len, out) ) {
            if (hash_size > sizeof(out)) {
                memset ((char*)hash_return + sizeof(out), 0, hash_size - sizeof(out));
                hash_size = sizeof(out);
            }
            memcpy(hash_return, out, hash_size);
        }
        free (in);
    }
}

int dht_random_bytes (void *buf, size_t size)
{
    int fd, rc = 0, save;
    size_t len = 0;

    fd = open("/dev/urandom", O_RDONLY);
    if (fd < 0) {
        return fd;
    }

    while (len < size) {
        rc = read (fd, buf + len, size - len);
        if (rc < 0) {
            if (errno == EINTR || errno == EAGAIN) {
                continue; // not fatal, try again.
            }
            if (errno == EWOULDBLOCK && len > 0) {
                rc = len;
            }

            break; // fail.
        }
        len += rc;
    }

    if (len > 0 && rc >= 0) {
        rc = len;
    }

    save = errno;
    close(fd);
    errno = save;

    return rc;
}