FS-9775: Changed local nodeid to be per endpoint, and removed ks_dht_node_t in…

FS-9775: Changed local nodeid to be per endpoint, and removed ks_dht_node_t in preparation for changes based on route table discussions

libs/libks/Makefile.am

@@ -13,7 +13,7 @@ libks_la_SOURCES += src/ks_time.c src/ks_printf.c src/ks_hash.c src/ks_q.c src/k
 libks_la_SOURCES += src/ks_ssl.c src/kws.c src/ks_rng.c
 libks_la_SOURCES += src/utp/utp_api.cpp src/utp/utp_callbacks.cpp src/utp/utp_hash.cpp src/utp/utp_internal.cpp
 libks_la_SOURCES += src/utp/utp_packedsockaddr.cpp src/utp/utp_utils.cpp src/ks_bencode.c
-libks_la_SOURCES += src/dht/ks_dht.c src/dht/ks_dht_endpoint.c src/dht/ks_dht_node.c src/dht/ks_dht_message.c src/dht/ks_dht_transaction.c
+libks_la_SOURCES += src/dht/ks_dht.c src/dht/ks_dht_endpoint.c src/dht/ks_dht_message.c src/dht/ks_dht_transaction.c
 libks_la_SOURCES += src/dht/ks_dht_bucket.c 
 libks_la_SOURCES += crypt/aeskey.c crypt/aestab.c crypt/sha2.c crypt/twofish.c crypt/aes_modes.c crypt/aescrypt.c crypt/twofish_cfb.c 
 #aes.h aescpp.h brg_endian.h aesopt.h aestab.h brg_types.h sha2.h twofish.h

libs/libks/src/dht/ks_dht-int.h

@@ -27,15 +27,16 @@ KS_DECLARE(void) ks_dht_idle_send(ks_dht_t *dht);
 
 KS_DECLARE(ks_status_t) ks_dht_send(ks_dht_t *dht, ks_dht_message_t *message);
 KS_DECLARE(ks_status_t) ks_dht_send_error(ks_dht_t *dht,
+										  ks_dht_endpoint_t *ep,
 										  ks_sockaddr_t *raddr,
 										  uint8_t *transactionid,
 										  ks_size_t transactionid_length,
 										  long long errorcode,
 										  const char *errorstr);
-KS_DECLARE(ks_status_t) ks_dht_send_ping(ks_dht_t *dht, ks_sockaddr_t *raddr);
-KS_DECLARE(ks_status_t) ks_dht_send_findnode(ks_dht_t *dht, ks_sockaddr_t *raddr, ks_dht_nodeid_t *targetid);
+KS_DECLARE(ks_status_t) ks_dht_send_ping(ks_dht_t *dht, ks_dht_endpoint_t *ep, ks_sockaddr_t *raddr);
+KS_DECLARE(ks_status_t) ks_dht_send_findnode(ks_dht_t *dht, ks_dht_endpoint_t *ep, ks_sockaddr_t *raddr, ks_dht_nodeid_t *targetid);
 
-KS_DECLARE(ks_status_t) ks_dht_process(ks_dht_t *dht, ks_sockaddr_t *raddr);
+KS_DECLARE(ks_status_t) ks_dht_process(ks_dht_t *dht, ks_dht_endpoint_t *ep, ks_sockaddr_t *raddr);
 
 KS_DECLARE(ks_status_t) ks_dht_process_query(ks_dht_t *dht, ks_dht_message_t *message);
 KS_DECLARE(ks_status_t) ks_dht_process_response(ks_dht_t *dht, ks_dht_message_t *message);
@@ -54,18 +55,25 @@ KS_DECLARE(ks_status_t) ks_dht_endpoint_alloc(ks_dht_endpoint_t **endpoint, ks_p
 KS_DECLARE(ks_status_t) ks_dht_endpoint_prealloc(ks_dht_endpoint_t *endpoint, ks_pool_t *pool);
 KS_DECLARE(ks_status_t) ks_dht_endpoint_free(ks_dht_endpoint_t *endpoint);
 
-KS_DECLARE(ks_status_t) ks_dht_endpoint_init(ks_dht_endpoint_t *endpoint, const ks_sockaddr_t *addr, ks_socket_t sock);
+KS_DECLARE(ks_status_t) ks_dht_endpoint_init(ks_dht_endpoint_t *endpoint,
+											 const ks_dht_nodeid_t *nodeid,
+											 const ks_sockaddr_t *addr,
+											 ks_socket_t sock);
 KS_DECLARE(ks_status_t) ks_dht_endpoint_deinit(ks_dht_endpoint_t *endpoint);
 
 /**
  *
  */
-KS_DECLARE(ks_status_t) ks_dht_node_alloc(ks_dht_node_t **node, ks_pool_t *pool);
-KS_DECLARE(ks_status_t) ks_dht_node_prealloc(ks_dht_node_t *node, ks_pool_t *pool);
-KS_DECLARE(ks_status_t) ks_dht_node_free(ks_dht_node_t *node);
+//KS_DECLARE(ks_status_t) ks_dht_node_alloc(ks_dht_node_t **node, ks_pool_t *pool);
+//KS_DECLARE(ks_status_t) ks_dht_node_prealloc(ks_dht_node_t *node, ks_pool_t *pool);
+//KS_DECLARE(ks_status_t) ks_dht_node_free(ks_dht_node_t *node);
 
-KS_DECLARE(ks_status_t) ks_dht_node_init(ks_dht_node_t *node, const ks_dht_nodeid_t *id, const ks_sockaddr_t *addr);
-KS_DECLARE(ks_status_t) ks_dht_node_deinit(ks_dht_node_t *node);
+//KS_DECLARE(ks_status_t) ks_dht_node_init(ks_dht_node_t *node, const ks_dht_nodeid_t *id, const ks_sockaddr_t *addr);
+//KS_DECLARE(ks_status_t) ks_dht_node_deinit(ks_dht_node_t *node);
+
+//KS_DECLARE(ks_status_t) ks_dht_node_address_check(ks_dht_node_t *node, const ks_sockaddr_t *addr);
+//KS_DECLARE(ks_bool_t) ks_dht_node_address_exists(ks_dht_node_t *node, const ks_sockaddr_t *addr);
+//KS_DECLARE(ks_status_t) ks_dht_node_address_add(ks_dht_node_t *node, const ks_sockaddr_t *addr);
 
 
 KS_END_EXTERN_C

libs/libks/src/dht/ks_dht.c

@@ -18,6 +18,23 @@ KS_DECLARE(ks_status_t) ks_dht_alloc(ks_dht_t **dht, ks_pool_t *pool)
 	d->pool = pool;
 	d->pool_alloc = pool_alloc;
 
+	d->autoroute = KS_FALSE;
+	d->autoroute_port = 0;
+	d->registry_type = NULL;
+	d->registry_query = NULL;
+	d->registry_error = NULL;
+	d->bind_ipv4 = KS_FALSE;
+	d->bind_ipv6 = KS_FALSE;
+	d->endpoints = NULL;
+	d->endpoints_size = 0;
+	d->endpoints_hash = NULL;
+	d->endpoints_poll = NULL;
+	d->send_q = NULL;
+	d->send_q_unsent = NULL;
+	d->recv_buffer_length = 0;
+	d->transactionid_next = 0;
+	d->transactions_hash = NULL;
+
 	return KS_STATUS_SUCCESS;
 }
 
@@ -32,6 +49,23 @@ KS_DECLARE(ks_status_t) ks_dht_prealloc(ks_dht_t *dht, ks_pool_t *pool)
 	dht->pool = pool;
 	dht->pool_alloc = KS_FALSE;
 
+	dht->autoroute = KS_FALSE;
+	dht->autoroute_port = 0;
+	dht->registry_type = NULL;
+	dht->registry_query = NULL;
+	dht->registry_error = NULL;
+	dht->bind_ipv4 = KS_FALSE;
+	dht->bind_ipv6 = KS_FALSE;
+	dht->endpoints = NULL;
+	dht->endpoints_size = 0;
+	dht->endpoints_hash = NULL;
+	dht->endpoints_poll = NULL;
+	dht->send_q = NULL;
+	dht->send_q_unsent = NULL;
+	dht->recv_buffer_length = 0;
+	dht->transactionid_next = 0;
+	dht->transactions_hash = NULL;
+	
 	return KS_STATUS_SUCCESS;
 }
 
@@ -56,7 +90,7 @@ KS_DECLARE(ks_status_t) ks_dht_free(ks_dht_t *dht)
 /**
  *
  */
-KS_DECLARE(ks_status_t) ks_dht_init(ks_dht_t *dht, const ks_dht_nodeid_t *nodeid)
+KS_DECLARE(ks_status_t) ks_dht_init(ks_dht_t *dht)
 {
 	ks_assert(dht);
 	ks_assert(dht->pool);
@@ -64,12 +98,6 @@ KS_DECLARE(ks_status_t) ks_dht_init(ks_dht_t *dht, const ks_dht_nodeid_t *nodeid
 	dht->autoroute = KS_FALSE;
 	dht->autoroute_port = 0;
 	
-    if (!nodeid) {
-		randombytes_buf(dht->nodeid, KS_DHT_NODEID_SIZE);
-	} else {
-		memcpy(dht->nodeid, nodeid, KS_DHT_NODEID_SIZE);
-	}
-	
 	ks_hash_create(&dht->registry_type, KS_HASH_MODE_DEFAULT, KS_HASH_FLAG_RWLOCK | KS_HASH_FLAG_DUP_CHECK, dht->pool);
 	ks_dht_register_type(dht, "q", ks_dht_process_query);
 	ks_dht_register_type(dht, "r", ks_dht_process_response);
@@ -225,7 +253,7 @@ KS_DECLARE(ks_status_t) ks_dht_register_error(ks_dht_t *dht, const char *value,
 /**
  *
  */
-KS_DECLARE(ks_status_t) ks_dht_bind(ks_dht_t *dht, const ks_sockaddr_t *addr, ks_dht_endpoint_t **endpoint)
+KS_DECLARE(ks_status_t) ks_dht_bind(ks_dht_t *dht, const ks_dht_nodeid_t *nodeid, const ks_sockaddr_t *addr, ks_dht_endpoint_t **endpoint)
 {
 	ks_dht_endpoint_t *ep;
 	ks_socket_t sock;
@@ -259,7 +287,7 @@ KS_DECLARE(ks_status_t) ks_dht_bind(ks_dht_t *dht, const ks_sockaddr_t *addr, ks
 		return KS_STATUS_FAIL;
 	}
 	
-	if (ks_dht_endpoint_init(ep, addr, sock) != KS_STATUS_SUCCESS) {
+	if (ks_dht_endpoint_init(ep, nodeid, addr, sock) != KS_STATUS_SUCCESS) {
 		ks_dht_endpoint_free(ep);
 		ks_socket_close(&sock);
 		return KS_STATUS_FAIL;
@@ -315,7 +343,7 @@ KS_DECLARE(void) ks_dht_pulse(ks_dht_t *dht, int32_t timeout)
 			
 				raddr.family = dht->endpoints[i]->addr.family;
 				if (ks_socket_recvfrom(dht->endpoints_poll[i].fd, dht->recv_buffer, &dht->recv_buffer_length, &raddr) == KS_STATUS_SUCCESS) {
-					ks_dht_process(dht, &raddr);
+					ks_dht_process(dht, dht->endpoints[i], &raddr);
 				}
 			}
 		}
@@ -502,7 +530,7 @@ KS_DECLARE(ks_status_t) ks_dht_send(ks_dht_t *dht, ks_dht_message_t *message)
 	if (!(ep = ks_hash_search(dht->endpoints_hash, ip, KS_UNLOCKED)) && dht->autoroute) {
 		ks_sockaddr_t addr;
 		ks_addr_set(&addr, ip, dht->autoroute_port, message->raddr.family);
-		if (ks_dht_bind(dht, &addr, &ep) != KS_STATUS_SUCCESS) {
+		if (ks_dht_bind(dht, NULL, &addr, &ep) != KS_STATUS_SUCCESS) {
 			return KS_STATUS_FAIL;
 		}
 	}
@@ -524,6 +552,7 @@ KS_DECLARE(ks_status_t) ks_dht_send(ks_dht_t *dht, ks_dht_message_t *message)
  *
  */
 KS_DECLARE(ks_status_t) ks_dht_send_error(ks_dht_t *dht,
+										  ks_dht_endpoint_t *ep,
 										  ks_sockaddr_t *raddr,
 										  uint8_t *transactionid,
 										  ks_size_t transactionid_length,
@@ -543,7 +572,7 @@ KS_DECLARE(ks_status_t) ks_dht_send_error(ks_dht_t *dht,
 		return KS_STATUS_FAIL;
 	}
 
-	if (ks_dht_message_init(error, raddr, KS_TRUE) != KS_STATUS_SUCCESS) {
+	if (ks_dht_message_init(error, ep, raddr, KS_TRUE) != KS_STATUS_SUCCESS) {
 		goto done;
 	}
 
@@ -571,6 +600,7 @@ KS_DECLARE(ks_status_t) ks_dht_send_error(ks_dht_t *dht,
  *
  */
 KS_DECLARE(ks_status_t) ks_dht_setup_query(ks_dht_t *dht,
+										   ks_dht_endpoint_t *ep,
 										   ks_sockaddr_t *raddr,
 										   const char *query,
 										   ks_dht_message_callback_t callback,
@@ -605,7 +635,7 @@ KS_DECLARE(ks_status_t) ks_dht_setup_query(ks_dht_t *dht,
 		goto done;
 	}
 
-	if (ks_dht_message_init(msg, raddr, KS_TRUE) != KS_STATUS_SUCCESS) {
+	if (ks_dht_message_init(msg, ep, raddr, KS_TRUE) != KS_STATUS_SUCCESS) {
 	    goto done;
 	}
 
@@ -637,7 +667,7 @@ KS_DECLARE(ks_status_t) ks_dht_setup_query(ks_dht_t *dht,
 /**
  *
  */
-KS_DECLARE(ks_status_t) ks_dht_send_ping(ks_dht_t *dht, ks_sockaddr_t *raddr)
+KS_DECLARE(ks_status_t) ks_dht_send_ping(ks_dht_t *dht, ks_dht_endpoint_t *ep, ks_sockaddr_t *raddr)
 {
 	ks_dht_message_t *message = NULL;
 	struct bencode *a = NULL;
@@ -645,11 +675,11 @@ KS_DECLARE(ks_status_t) ks_dht_send_ping(ks_dht_t *dht, ks_sockaddr_t *raddr)
 	ks_assert(dht);
 	ks_assert(raddr);
 
-	if (ks_dht_setup_query(dht, raddr, "ping", ks_dht_process_response_ping, &message, &a) != KS_STATUS_SUCCESS) {
+	if (ks_dht_setup_query(dht, ep, raddr, "ping", ks_dht_process_response_ping, &message, &a) != KS_STATUS_SUCCESS) {
 		return KS_STATUS_FAIL;
 	}
 
-	ben_dict_set(a, ben_blob("id", 2), ben_blob(dht->nodeid, KS_DHT_NODEID_SIZE));
+	ben_dict_set(a, ben_blob("id", 2), ben_blob(ep->nodeid, KS_DHT_NODEID_SIZE));
 
 	ks_log(KS_LOG_DEBUG, "Sending message query ping\n");
 	ks_q_push(dht->send_q, (void *)message);
@@ -660,7 +690,7 @@ KS_DECLARE(ks_status_t) ks_dht_send_ping(ks_dht_t *dht, ks_sockaddr_t *raddr)
 /**
  *
  */
-KS_DECLARE(ks_status_t) ks_dht_send_findnode(ks_dht_t *dht, ks_sockaddr_t *raddr, ks_dht_nodeid_t *targetid)
+KS_DECLARE(ks_status_t) ks_dht_send_findnode(ks_dht_t *dht, ks_dht_endpoint_t *ep, ks_sockaddr_t *raddr, ks_dht_nodeid_t *targetid)
 {
 	ks_dht_message_t *message = NULL;
 	struct bencode *a = NULL;
@@ -669,11 +699,11 @@ KS_DECLARE(ks_status_t) ks_dht_send_findnode(ks_dht_t *dht, ks_sockaddr_t *raddr
 	ks_assert(raddr);
 	ks_assert(targetid);
 
-	if (ks_dht_setup_query(dht, raddr, "find_node", ks_dht_process_response_findnode, &message, &a) != KS_STATUS_SUCCESS) {
+	if (ks_dht_setup_query(dht, ep, raddr, "find_node", ks_dht_process_response_findnode, &message, &a) != KS_STATUS_SUCCESS) {
 		return KS_STATUS_FAIL;
 	}
 	
-	ben_dict_set(a, ben_blob("id", 2), ben_blob(dht->nodeid, KS_DHT_NODEID_SIZE));
+	ben_dict_set(a, ben_blob("id", 2), ben_blob(ep->nodeid, KS_DHT_NODEID_SIZE));
 	ben_dict_set(a, ben_blob("target", 6), ben_blob(targetid, KS_DHT_NODEID_SIZE));
 
 	ks_log(KS_LOG_DEBUG, "Sending message query find_node\n");
@@ -686,7 +716,7 @@ KS_DECLARE(ks_status_t) ks_dht_send_findnode(ks_dht_t *dht, ks_sockaddr_t *raddr
 /**
  *
  */
-KS_DECLARE(ks_status_t) ks_dht_process(ks_dht_t *dht, ks_sockaddr_t *raddr)
+KS_DECLARE(ks_status_t) ks_dht_process(ks_dht_t *dht, ks_dht_endpoint_t *ep, ks_sockaddr_t *raddr)
 {
 	ks_dht_message_t message;
 	ks_dht_message_callback_t callback;
@@ -707,7 +737,7 @@ KS_DECLARE(ks_status_t) ks_dht_process(ks_dht_t *dht, ks_sockaddr_t *raddr)
 		return KS_STATUS_FAIL;
 	}
 
-	if (ks_dht_message_init(&message, raddr, KS_FALSE) != KS_STATUS_SUCCESS) {
+	if (ks_dht_message_init(&message, ep, raddr, KS_FALSE) != KS_STATUS_SUCCESS) {
 		return KS_STATUS_FAIL;
 	}
 
@@ -938,7 +968,7 @@ KS_DECLARE(ks_status_t) ks_dht_process_query_ping(ks_dht_t *dht, ks_dht_message_
 		goto done;
 	}
 
-	if (ks_dht_message_init(response, &message->raddr, KS_TRUE) != KS_STATUS_SUCCESS) {
+	if (ks_dht_message_init(response, message->endpoint, &message->raddr, KS_TRUE) != KS_STATUS_SUCCESS) {
 		goto done;
 	}
 
@@ -946,7 +976,7 @@ KS_DECLARE(ks_status_t) ks_dht_process_query_ping(ks_dht_t *dht, ks_dht_message_
 		goto done;
 	}
 	
-	ben_dict_set(r, ben_blob("id", 2), ben_blob(dht->nodeid, KS_DHT_NODEID_SIZE));
+	ben_dict_set(r, ben_blob("id", 2), ben_blob(message->endpoint->nodeid, KS_DHT_NODEID_SIZE));
 
 	ks_log(KS_LOG_DEBUG, "Sending message response ping\n");
 	ks_q_push(dht->send_q, (void *)response);
@@ -1049,7 +1079,7 @@ KS_DECLARE(ks_status_t) ks_dht_process_query_findnode(ks_dht_t *dht, ks_dht_mess
 		goto done;
 	}
 
-	if (ks_dht_message_init(response, &message->raddr, KS_TRUE) != KS_STATUS_SUCCESS) {
+	if (ks_dht_message_init(response, message->endpoint, &message->raddr, KS_TRUE) != KS_STATUS_SUCCESS) {
 		goto done;
 	}
 
@@ -1057,7 +1087,7 @@ KS_DECLARE(ks_status_t) ks_dht_process_query_findnode(ks_dht_t *dht, ks_dht_mess
 		goto done;
 	}
 	
-	ben_dict_set(r, ben_blob("id", 2), ben_blob(dht->nodeid, KS_DHT_NODEID_SIZE));
+	ben_dict_set(r, ben_blob("id", 2), ben_blob(message->endpoint->nodeid, KS_DHT_NODEID_SIZE));
 	// @todo populate nodes/nodes6
 	ben_dict_set(r, ben_blob("nodes", 5), ben_blob(buffer, buffer_length));
 

libs/libks/src/dht/ks_dht.h

@@ -30,17 +30,10 @@ typedef struct ks_dht_transaction_s ks_dht_transaction_t;
 
 typedef ks_status_t (*ks_dht_message_callback_t)(ks_dht_t *dht, ks_dht_message_t *message);
 
-struct ks_dht_node_s {
-	ks_pool_t *pool;
-	ks_dht_nodeid_t id;
-	ks_sockaddr_t *addr4;
-	ks_sockaddr_t *addr6;
-	ks_size_t addr4_length;
-	ks_size_t addr6_length;
-};
 
 struct ks_dht_message_s {
 	ks_pool_t *pool;
+	ks_dht_endpoint_t *endpoint;
 	ks_sockaddr_t raddr;
 	struct bencode *data;
 	uint8_t transactionid[KS_DHT_MESSAGE_TRANSACTIONID_MAX_SIZE];
@@ -51,6 +44,7 @@ struct ks_dht_message_s {
 
 struct ks_dht_endpoint_s {
 	ks_pool_t *pool;
+	ks_dht_nodeid_t nodeid;
 	ks_sockaddr_t addr;
 	ks_socket_t sock;
 };
@@ -72,8 +66,6 @@ struct ks_dht_s {
 	ks_bool_t autoroute;
 	ks_port_t autoroute_port;
 	
-	ks_dht_nodeid_t nodeid;
-
 	ks_hash_t *registry_type;
 	ks_hash_t *registry_query;
 	ks_hash_t *registry_error;
@@ -103,12 +95,12 @@ KS_DECLARE(ks_status_t) ks_dht_prealloc(ks_dht_t *dht, ks_pool_t *pool);
 KS_DECLARE(ks_status_t) ks_dht_free(ks_dht_t *dht);
 
 
-KS_DECLARE(ks_status_t) ks_dht_init(ks_dht_t *dht, const ks_dht_nodeid_t *nodeid);
+KS_DECLARE(ks_status_t) ks_dht_init(ks_dht_t *dht);
 KS_DECLARE(ks_status_t) ks_dht_deinit(ks_dht_t *dht);
 
 KS_DECLARE(ks_status_t) ks_dht_autoroute(ks_dht_t *dht, ks_bool_t autoroute, ks_port_t port);
 
-KS_DECLARE(ks_status_t) ks_dht_bind(ks_dht_t *dht, const ks_sockaddr_t *addr, ks_dht_endpoint_t **endpoint);
+KS_DECLARE(ks_status_t) ks_dht_bind(ks_dht_t *dht, const ks_dht_nodeid_t *nodeid, const ks_sockaddr_t *addr, ks_dht_endpoint_t **endpoint);
 KS_DECLARE(void) ks_dht_pulse(ks_dht_t *dht, int32_t timeout);
 
 
@@ -122,7 +114,7 @@ KS_DECLARE(ks_status_t) ks_dht_message_alloc(ks_dht_message_t **message, ks_pool
 KS_DECLARE(ks_status_t) ks_dht_message_prealloc(ks_dht_message_t *message, ks_pool_t *pool);
 KS_DECLARE(ks_status_t) ks_dht_message_free(ks_dht_message_t *message);
 
-KS_DECLARE(ks_status_t) ks_dht_message_init(ks_dht_message_t *message, ks_sockaddr_t *raddr, ks_bool_t alloc_data);
+KS_DECLARE(ks_status_t) ks_dht_message_init(ks_dht_message_t *message, ks_dht_endpoint_t *ep, ks_sockaddr_t *raddr, ks_bool_t alloc_data);
 KS_DECLARE(ks_status_t) ks_dht_message_deinit(ks_dht_message_t *message);
 
 KS_DECLARE(ks_status_t) ks_dht_message_parse(ks_dht_message_t *message, const uint8_t *buffer, ks_size_t buffer_length);

libs/libks/src/dht/ks_dht_endpoint.c

 #include "ks_dht.h"
 #include "ks_dht-int.h"
+#include "sodium.h"
 
 /**
  *
@@ -49,13 +50,19 @@ KS_DECLARE(ks_status_t) ks_dht_endpoint_free(ks_dht_endpoint_t *endpoint)
 /**
  *
  */
-KS_DECLARE(ks_status_t) ks_dht_endpoint_init(ks_dht_endpoint_t *endpoint, const ks_sockaddr_t *addr, ks_socket_t sock)
+KS_DECLARE(ks_status_t) ks_dht_endpoint_init(ks_dht_endpoint_t *endpoint, const ks_dht_nodeid_t *nodeid, const ks_sockaddr_t *addr, ks_socket_t sock)
 {
 	ks_assert(endpoint);
 	ks_assert(endpoint->pool);
 	ks_assert(addr);
 	ks_assert(addr->family == AF_INET || addr->family == AF_INET6);
 	
+    if (!nodeid) {
+		randombytes_buf(endpoint->nodeid, KS_DHT_NODEID_SIZE);
+	} else {
+		memcpy(endpoint->nodeid, nodeid, KS_DHT_NODEID_SIZE);
+	}
+
 	endpoint->addr = *addr;
 	endpoint->sock = sock;
 

libs/libks/src/dht/ks_dht_message.c

@@ -13,6 +13,12 @@ KS_DECLARE(ks_status_t) ks_dht_message_alloc(ks_dht_message_t **message, ks_pool
 	
 	*message = msg = ks_pool_alloc(pool, sizeof(ks_dht_message_t));
 	msg->pool = pool;
+	msg->endpoint = NULL;
+	msg->raddr = (const ks_sockaddr_t){ 0 };
+	msg->args = NULL;
+	msg->type[0] = '\0';
+	msg->transactionid_length = 0;
+	msg->data = NULL;
 
 	return KS_STATUS_SUCCESS;
 }
@@ -26,6 +32,12 @@ KS_DECLARE(ks_status_t) ks_dht_message_prealloc(ks_dht_message_t *message, ks_po
 	ks_assert(pool);
 	
 	message->pool = pool;
+	message->endpoint = NULL;
+	message->raddr = (const ks_sockaddr_t){ 0 };
+	message->args = NULL;
+	message->type[0] = '\0';
+	message->transactionid_length = 0;
+	message->data = NULL;
 
 	return KS_STATUS_SUCCESS;
 }
@@ -47,11 +59,12 @@ KS_DECLARE(ks_status_t) ks_dht_message_free(ks_dht_message_t *message)
 /**
  *
  */
-KS_DECLARE(ks_status_t) ks_dht_message_init(ks_dht_message_t *message, ks_sockaddr_t *raddr, ks_bool_t alloc_data)
+KS_DECLARE(ks_status_t) ks_dht_message_init(ks_dht_message_t *message, ks_dht_endpoint_t *ep, ks_sockaddr_t *raddr, ks_bool_t alloc_data)
 {
 	ks_assert(message);
 	ks_assert(message->pool);
 
+	message->endpoint = ep;
 	message->raddr = *raddr;
 	message->data = NULL;
 	message->args = NULL;
@@ -71,6 +84,7 @@ KS_DECLARE(ks_status_t) ks_dht_message_deinit(ks_dht_message_t *message)
 {
 	ks_assert(message);
 
+	message->endpoint = NULL;
 	message->raddr = (const ks_sockaddr_t){ 0 };
 	message->args = NULL;
 	message->type[0] = '\0';

libs/libks/src/dht/ks_dht_node.c

-#include "ks_dht.h"
-#include "ks_dht-int.h"
-#include "sodium.h"
-
-/**
- *
- */
-KS_DECLARE(ks_status_t) ks_dht_node_alloc(ks_dht_node_t **node, ks_pool_t *pool)
-{
-	ks_dht_node_t *n;
-
-	ks_assert(node);
-	ks_assert(pool);
-
-	*node = n = ks_pool_alloc(pool, sizeof(ks_dht_node_t));
-	n->pool = pool;
-
-	return KS_STATUS_SUCCESS;
-}
-
-/**
- *
- */
-KS_DECLARE(ks_status_t) ks_dht_node_prealloc(ks_dht_node_t *node, ks_pool_t *pool)
-{
-	ks_assert(node);
-	ks_assert(pool);
-	
-	node->pool = pool;
-
-	return KS_STATUS_SUCCESS;
-}
-
-/**
- *
- */
-KS_DECLARE(ks_status_t) ks_dht_node_free(ks_dht_node_t *node)
-{
-	ks_assert(node);
-
-	ks_dht_node_deinit(node);
-	ks_pool_free(node->pool, node);
-
-	return KS_STATUS_SUCCESS;
-}
-
-
-/**
- *
- */
-KS_DECLARE(ks_status_t) ks_dht_node_init(ks_dht_node_t *node, const ks_dht_nodeid_t *id, const ks_sockaddr_t *addr)
-{
-	ks_assert(node);
-	ks_assert(node->pool);
-	ks_assert(id);
-	ks_assert(addr);
-
-	return KS_STATUS_SUCCESS;
-}
-
-/**
- *
- */
-KS_DECLARE(ks_status_t) ks_dht_node_deinit(ks_dht_node_t *node)
-{
-	ks_assert(node);
-
-	return KS_STATUS_SUCCESS;
-}
-
-
-/* For Emacs:
- * Local Variables:
- * mode:c
- * indent-tabs-mode:t
- * tab-width:4
- * c-basic-offset:4
- * End:
- * For VIM:
- * vim:set softtabstop=4 shiftwidth=4 tabstop=4 noet:
- */

libs/libks/src/dht/ks_dht_transaction.c

@@ -13,6 +13,11 @@ KS_DECLARE(ks_status_t) ks_dht_transaction_alloc(ks_dht_transaction_t **transact
 	
 	*transaction = tran = ks_pool_alloc(pool, sizeof(ks_dht_transaction_t));
 	tran->pool = pool;
+	tran->raddr = (const ks_sockaddr_t){ 0 };
+	tran->transactionid = 0;
+	tran->callback = NULL;
+	tran->expiration = 0;
+	tran->finished = KS_FALSE;
 
 	return KS_STATUS_SUCCESS;
 }
@@ -26,6 +31,11 @@ KS_DECLARE(ks_status_t) ks_dht_transaction_prealloc(ks_dht_transaction_t *transa
 	ks_assert(pool);
 	
 	transaction->pool = pool;
+	transaction->raddr = (const ks_sockaddr_t){ 0 };
+	transaction->transactionid = 0;
+	transaction->callback = NULL;
+	transaction->expiration = 0;
+	transaction->finished = KS_FALSE;
 
 	return KS_STATUS_SUCCESS;
 }

libs/libks/test/testdht2.c

@@ -10,7 +10,7 @@ ks_status_t dht_z_callback(ks_dht_t *dht, ks_dht_message_t *message)
 {
 	diag("dht_z_callback\n");
 	ok(message->transactionid[0] == '4' && message->transactionid[1] == '2');
-	ks_dht_send_error(dht, &message->raddr, message->transactionid, message->transactionid_length, 201, "Generic test error");
+	ks_dht_send_error(dht, message->endpoint, &message->raddr, message->transactionid, message->transactionid_length, 201, "Generic test error");
 	return KS_STATUS_SUCCESS;
 }
 
@@ -20,6 +20,8 @@ int main() {
   int mask = 0;
   ks_dht_t *dht1 = NULL;
   ks_dht_t dht2;
+  ks_dht_endpoint_t *ep1;
+  ks_dht_endpoint_t *ep2;
   ks_bool_t have_v4, have_v6;
   char v4[48] = {0}, v6[48] = {0};
   ks_sockaddr_t addr;
@@ -50,13 +52,13 @@ int main() {
   err = ks_dht_alloc(&dht1, NULL);
   ok(err == KS_STATUS_SUCCESS);
   
-  err = ks_dht_init(dht1, NULL);
+  err = ks_dht_init(dht1);
   ok(err == KS_STATUS_SUCCESS);
 
   err = ks_dht_prealloc(&dht2, dht1->pool);
   ok(err == KS_STATUS_SUCCESS);
   
-  err = ks_dht_init(&dht2, NULL);
+  err = ks_dht_init(&dht2);
   ok(err == KS_STATUS_SUCCESS);
 
   ks_dht_register_type(dht1, "z", dht_z_callback);
@@ -65,13 +67,13 @@ int main() {
     err = ks_addr_set(&addr, v4, KS_DHT_DEFAULT_PORT, AF_INET);
 	ok(err == KS_STATUS_SUCCESS);
 	
-    err = ks_dht_bind(dht1, &addr, NULL);
+    err = ks_dht_bind(dht1, NULL, &addr, &ep1);
     ok(err == KS_STATUS_SUCCESS);
 
 	err = ks_addr_set(&addr, v4, KS_DHT_DEFAULT_PORT + 1, AF_INET);
 	ok(err == KS_STATUS_SUCCESS);
 	
-	err = ks_dht_bind(&dht2, &addr, NULL);
+	err = ks_dht_bind(&dht2, NULL, &addr, &ep2);
 	ok(err == KS_STATUS_SUCCESS);
 
 	raddr = addr;
@@ -81,13 +83,13 @@ int main() {
 	err = ks_addr_set(&addr, v6, KS_DHT_DEFAULT_PORT, AF_INET6);
 	ok(err == KS_STATUS_SUCCESS);
 	  
-    err = ks_dht_bind(dht1, &addr, NULL);
+    err = ks_dht_bind(dht1, NULL, &addr, NULL);
     ok(err == KS_STATUS_SUCCESS);
 
 	err = ks_addr_set(&addr, v6, KS_DHT_DEFAULT_PORT + 1, AF_INET6);
 	ok(err == KS_STATUS_SUCCESS);
 
-	err = ks_dht_bind(&dht2, &addr, NULL);
+	err = ks_dht_bind(&dht2, NULL, &addr, NULL);
 	ok(err == KS_STATUS_SUCCESS);
   }
 
@@ -97,7 +99,7 @@ int main() {
   memcpy(dht1->recv_buffer, TEST_DHT1_REGISTER_TYPE_BUFFER, buflen);
   dht1->recv_buffer_length = buflen;
 
-  err = ks_dht_process(dht1, &raddr);
+  err = ks_dht_process(dht1, ep1, &raddr);
   ok(err == KS_STATUS_SUCCESS);
 
   ks_dht_pulse(dht1, 100);
@@ -115,7 +117,7 @@ int main() {
   
   diag("Ping tests\n");
   
-  ks_dht_send_ping(dht1, &raddr);
+  ks_dht_send_ping(dht1, ep1, &raddr);
 
   ks_dht_pulse(dht1, 100);