From: Guus Sliepen <guus@meshlink.io>
Date: Sun, 12 Aug 2018 15:09:20 +0000 (+0200)
Subject: Add duplicate node detection callback.
X-Git-Url: https://git.meshlink.io/?a=commitdiff_plain;h=f712fdc0b3de22566cdf06954256c62f46f25542;p=meshlink

Add duplicate node detection callback.

MeshLink can detect if two nodes are online simultaneously using the same
Name. Normally, one or both of the duplicate nodes will terminate its
connection. Now there is a duplicate node callback that will be called when
the local node detects a duplicate node. One use for this is to blacklist
the duplicate node and/or notify the user of the problem.
---

diff --git a/src/graph.c b/src/graph.c
index a6e38e42..9b0184b0 100644
--- a/src/graph.c
+++ b/src/graph.c
@@ -233,7 +233,7 @@ static void check_reachability(meshlink_handle_t *mesh) {
 
 			if(!n->status.reachable) {
 				update_node_udp(mesh, n, NULL);
-				memset(&n->status, 0, sizeof(n)->status);
+				n->status.broadcast = false;
 				n->options = 0;
 			} else if(n->connection) {
 				if(n->connection->outgoing) {
diff --git a/src/meshlink++.h b/src/meshlink++.h
index 413534ea..4179a9ec 100644
--- a/src/meshlink++.h
+++ b/src/meshlink++.h
@@ -49,6 +49,13 @@ typedef void (*receive_cb_t)(mesh *mesh, node *source, const void *data, size_t
  */
 typedef void (*node_status_cb_t)(mesh *mesh, node *node, bool reachable);
 
+/// A callback reporting duplicate node detection.
+/** @param mesh       A handle which represents an instance of MeshLink.
+ *  @param node       A pointer to a meshlink_node_t describing the node which is duplicate.
+ *                    This pointer is valid until meshlink_close() is called.
+ */
+typedef void (*duplicate_cb_t)(mesh *mesh, node *node);
+
 /// A callback for receiving log messages generated by MeshLink.
 /** @param mesh      A handle which represents an instance of MeshLink.
  *  @param level     An enum describing the severity level of the message.
@@ -181,15 +188,21 @@ public:
 		(void) length;
 	}
 
-	/// This functions is called  whenever another node's status changed.
-	virtual void node_status(node *peer, bool reachable)                {
+	/// This functions is called whenever another node's status changed.
+	virtual void node_status(node *peer, bool reachable) {
 		/* do nothing */
 		(void)peer;
 		(void)reachable;
 	}
 
+	/// This functions is called whenever a duplicate node is detected.
+	virtual void node_duplicate(node *peer) {
+		/* do nothing */
+		(void)peer;
+	}
+
 	/// This functions is called whenever MeshLink has some information to log.
-	virtual void log(log_level_t level, const char *message)            {
+	virtual void log(log_level_t level, const char *message) {
 		/* do nothing */
 		(void)level;
 		(void)message;
@@ -266,6 +279,7 @@ public:
 	bool start() {
 		meshlink_set_receive_cb(handle, &receive_trampoline);
 		meshlink_set_node_status_cb(handle, &node_status_trampoline);
+		meshlink_set_node_duplicate_cb(handle, &node_duplicate_trampoline);
 		meshlink_set_log_cb(handle, MESHLINK_DEBUG, &log_trampoline);
 		meshlink_set_channel_accept_cb(handle, &channel_accept_trampoline);
 		return meshlink_start(handle);
@@ -670,6 +684,15 @@ private:
 		that->node_status(static_cast<node *>(peer), reachable);
 	}
 
+	static void node_duplicate_trampoline(meshlink_handle_t *handle, meshlink_node_t *peer) {
+		if(!(handle->priv)) {
+			return;
+		}
+
+		meshlink::mesh *that = static_cast<mesh *>(handle->priv);
+		that->node_duplicate(static_cast<node *>(peer));
+	}
+
 	static void log_trampoline(meshlink_handle_t *handle, log_level_t level, const char *message) {
 		if(!(handle->priv)) {
 			return;
diff --git a/src/meshlink.c b/src/meshlink.c
index c720c766..69dd7141 100644
--- a/src/meshlink.c
+++ b/src/meshlink.c
@@ -1216,9 +1216,11 @@ bool meshlink_start(meshlink_handle_t *mesh) {
 	mesh->threadstarted = true;
 
 #if HAVE_CATTA
+
 	if(mesh->discovery) {
 		discovery_start(mesh);
 	}
+
 #endif
 
 	pthread_mutex_unlock(&(mesh->mesh_mutex));
@@ -1235,10 +1237,12 @@ void meshlink_stop(meshlink_handle_t *mesh) {
 	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
@@ -1374,6 +1378,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_duplicate_cb(meshlink_handle_t *mesh, meshlink_node_duplicate_cb_t cb) {
+	if(!mesh) {
+		meshlink_errno = MESHLINK_EINVAL;
+		return;
+	}
+
+	pthread_mutex_lock(&(mesh->mesh_mutex));
+	mesh->node_duplicate_cb = cb;
+	pthread_mutex_unlock(&(mesh->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));
@@ -2365,8 +2380,14 @@ void meshlink_blacklist(meshlink_handle_t *mesh, meshlink_node_t *node) {
 	//Make blacklisting persistent in the config file
 	append_config_file(mesh, n->name, "blacklisted", "yes");
 
+	//Immediately terminate any connections we have with the blacklisted node
+	for list_each(connection_t, c, mesh->connections) {
+		if(c->node == n) {
+			terminate_connection(mesh, c, c->status.active);
+		}
+	}
+
 	pthread_mutex_unlock(&(mesh->mesh_mutex));
-	return;
 }
 
 void meshlink_whitelist(meshlink_handle_t *mesh, meshlink_node_t *node) {
@@ -2656,8 +2677,18 @@ void update_node_status(meshlink_handle_t *mesh, node_t *n) {
 	}
 }
 
+void handle_duplicate_node(meshlink_handle_t *mesh, node_t *n) {
+	if(!mesh->node_duplicate_cb || n->status.duplicate) {
+		return;
+	}
+
+	n->status.duplicate = true;
+	mesh->node_duplicate_cb(mesh, (meshlink_node_t *)n);
+}
+
 void meshlink_enable_discovery(meshlink_handle_t *mesh, bool enable) {
 #if HAVE_CATTA
+
 	if(!mesh) {
 		meshlink_errno = MESHLINK_EINVAL;
 		return;
diff --git a/src/meshlink.h b/src/meshlink.h
index 37eeac81..15fb6e69 100644
--- a/src/meshlink.h
+++ b/src/meshlink.h
@@ -242,6 +242,26 @@ typedef void (*meshlink_node_status_cb_t)(meshlink_handle_t *mesh, meshlink_node
  */
 extern void meshlink_set_node_status_cb(meshlink_handle_t *mesh, meshlink_node_status_cb_t cb);
 
+/// A callback reporting duplicate node detection.
+/** @param mesh       A handle which represents an instance of MeshLink.
+ *  @param node       A pointer to a meshlink_node_t describing the node which is duplicate.
+ *                    This pointer is valid until meshlink_close() is called.
+ */
+typedef void (*meshlink_node_duplicate_cb_t)(meshlink_handle_t *mesh, meshlink_node_t *node);
+
+/// Set the node duplicate callback.
+/** This functions sets the callback that is called whenever a duplicate node is detected.
+ *  The callback is run in MeshLink's own thread.
+ *  It is therefore important that the callback uses apprioriate methods (queues, pipes, locking, etc.)
+ *  to hand the data over to the application's thread.
+ *  The callback should also not block itself and return as quickly as possible.
+ *
+ *  @param mesh      A handle which represents an instance of MeshLink.
+ *  @param cb        A pointer to the function which will be called when a duplicate node is detected.
+ *                   If a NULL pointer is given, the callback will be disabled.
+ */
+extern void meshlink_set_node_duplicate_cb(meshlink_handle_t *mesh, meshlink_node_duplicate_cb_t cb);
+
 /// Severity of log messages generated by MeshLink.
 typedef enum {
 	MESHLINK_DEBUG,    ///< Internal debugging messages. Only useful during application development.
diff --git a/src/meshlink_internal.h b/src/meshlink_internal.h
index c6c99541..545bd526 100644
--- a/src/meshlink_internal.h
+++ b/src/meshlink_internal.h
@@ -73,6 +73,7 @@ struct meshlink_handle {
 	meshlink_log_level_t log_level;
 
 	meshlink_channel_accept_cb_t channel_accept_cb;
+	meshlink_node_duplicate_cb_t node_duplicate_cb;
 
 	pthread_t thread;
 	bool threadstarted;
@@ -172,6 +173,7 @@ extern void update_node_status(meshlink_handle_t *mesh, struct node_t *n);
 extern meshlink_log_level_t global_log_level;
 extern meshlink_log_cb_t global_log_cb;
 extern int check_port(meshlink_handle_t *mesh);
+extern void handle_duplicate_node(meshlink_handle_t *mesh, struct node_t *n);
 
 /// Device class traits
 typedef struct {
diff --git a/src/net.c b/src/net.c
index 84625917..9d5e8171 100644
--- a/src/net.c
+++ b/src/net.c
@@ -57,7 +57,7 @@ void terminate_connection(meshlink_handle_t *mesh, connection_t *c, bool report)
 
 	if(c->edge) {
 		if(report) {
-			send_del_edge(mesh, mesh->everyone, c->edge);
+			send_del_edge(mesh, mesh->everyone, c->edge, 0);
 		}
 
 		edge_del(mesh, c->edge);
@@ -74,7 +74,7 @@ void terminate_connection(meshlink_handle_t *mesh, connection_t *c, bool report)
 			e = lookup_edge(c->node, mesh->self);
 
 			if(e) {
-				send_del_edge(mesh, mesh->everyone, e);
+				send_del_edge(mesh, mesh->everyone, e, 0);
 				edge_del(mesh, e);
 			}
 		}
diff --git a/src/node.h b/src/node.h
index ba5b1d48..eaba79d0 100644
--- a/src/node.h
+++ b/src/node.h
@@ -37,7 +37,8 @@ typedef struct node_status_t {
 	unsigned int broadcast: 1;              /* 1 if the next UDP packet should be broadcast to the local network */
 	unsigned int blacklisted: 1;            /* 1 if the node is blacklist so we never want to speak with him anymore */
 	unsigned int destroyed: 1;              /* 1 if the node is being destroyed, deallocate channels when any callback is triggered */
-	unsigned int unused: 22;
+	unsigned int duplicate: 1;              /* 1 if the node is duplicate, ie. multiple nodes using the same Name are online */
+	unsigned int unused: 20;
 } node_status_t;
 
 typedef struct node_t {
diff --git a/src/protocol.h b/src/protocol.h
index 48bdd864..92ccef01 100644
--- a/src/protocol.h
+++ b/src/protocol.h
@@ -85,8 +85,8 @@ extern bool send_id(struct meshlink_handle *mesh, struct connection_t *);
 extern bool send_ack(struct meshlink_handle *mesh, struct connection_t *);
 extern bool send_ping(struct meshlink_handle *mesh, struct connection_t *);
 extern bool send_pong(struct meshlink_handle *mesh, struct connection_t *);
-extern bool send_add_edge(struct meshlink_handle *mesh, struct connection_t *, const struct edge_t *);
-extern bool send_del_edge(struct meshlink_handle *mesh, struct connection_t *, const struct edge_t *);
+extern bool send_add_edge(struct meshlink_handle *mesh, struct connection_t *, const struct edge_t *, int contradictions);
+extern bool send_del_edge(struct meshlink_handle *mesh, struct connection_t *, const struct edge_t *, int contradictions);
 extern void send_key_changed(struct meshlink_handle *mesh);
 extern bool send_req_key(struct meshlink_handle *mesh, struct node_t *);
 
diff --git a/src/protocol_auth.c b/src/protocol_auth.c
index b43003e3..e56aa881 100644
--- a/src/protocol_auth.c
+++ b/src/protocol_auth.c
@@ -387,6 +387,14 @@ bool id_h(meshlink_handle_t *mesh, connection_t *c, const char *request) {
 		}
 	}
 
+	bool blacklisted = false;
+	get_config_bool(lookup_config(c->config_tree, "blacklisted"), &blacklisted);
+
+	if(blacklisted) {
+		logger(mesh, MESHLINK_EPEER, "Peer %s is blacklisted", c->name);
+		return false;
+	}
+
 	read_ecdsa_public_key(mesh, c);
 
 	if(!ecdsa_active(c->ecdsa)) {
@@ -438,7 +446,7 @@ static void send_everything(meshlink_handle_t *mesh, connection_t *c) {
 
 	for splay_each(node_t, n, mesh->nodes) {
 		for splay_each(edge_t, e, n->edge_tree) {
-			send_add_edge(mesh, c, e);
+			send_add_edge(mesh, c, e, 0);
 		}
 	}
 }
@@ -533,7 +541,7 @@ bool ack_h(meshlink_handle_t *mesh, connection_t *c, const char *request) {
 
 	/* Notify everyone of the new edge */
 
-	send_add_edge(mesh, mesh->everyone, c->edge);
+	send_add_edge(mesh, mesh->everyone, c->edge, 0);
 
 	/* Run MST and SSSP algorithms */
 
diff --git a/src/protocol_edge.c b/src/protocol_edge.c
index 40172b9d..67881098 100644
--- a/src/protocol_edge.c
+++ b/src/protocol_edge.c
@@ -35,15 +35,15 @@
 
 extern bool node_write_devclass(meshlink_handle_t *mesh, node_t *n);
 
-bool send_add_edge(meshlink_handle_t *mesh, connection_t *c, const edge_t *e) {
+bool send_add_edge(meshlink_handle_t *mesh, connection_t *c, const edge_t *e, int contradictions) {
 	bool x;
 	char *address, *port;
 
 	sockaddr2str(&e->address, &address, &port);
 
-	x = send_request(mesh, c, "%d %x %s %d %s %s %s %d %x %d", ADD_EDGE, rand(),
+	x = send_request(mesh, c, "%d %x %s %d %s %s %s %d %x %d %d", ADD_EDGE, rand(),
 	                 e->from->name, e->from->devclass, e->to->name, address, port, e->to->devclass,
-	                 e->options, e->weight);
+	                 e->options, e->weight, contradictions);
 	free(address);
 	free(port);
 
@@ -62,9 +62,10 @@ bool add_edge_h(meshlink_handle_t *mesh, connection_t *c, const char *request) {
 	sockaddr_t address;
 	uint32_t options;
 	int weight;
+	int contradictions = 0;
 
-	if(sscanf(request, "%*d %*x "MAX_STRING" %d "MAX_STRING" "MAX_STRING" "MAX_STRING" %d %x %d",
-	                from_name, &from_devclass, to_name, to_address, to_port, &to_devclass, &options, &weight) != 8) {
+	if(sscanf(request, "%*d %*x "MAX_STRING" %d "MAX_STRING" "MAX_STRING" "MAX_STRING" %d %x %d %d",
+	                from_name, &from_devclass, to_name, to_address, to_port, &to_devclass, &options, &weight, &contradictions) < 8) {
 		logger(mesh, MESHLINK_ERROR, "Got bad %s from %s", "ADD_EDGE", c->name);
 		return false;
 	}
@@ -104,6 +105,10 @@ bool add_edge_h(meshlink_handle_t *mesh, connection_t *c, const char *request) {
 		node_add(mesh, from);
 	}
 
+	if(contradictions > 50) {
+		handle_duplicate_node(mesh, from);
+	}
+
 	from->devclass = from_devclass;
 	node_write_devclass(mesh, from);
 
@@ -130,7 +135,7 @@ bool add_edge_h(meshlink_handle_t *mesh, connection_t *c, const char *request) {
 			if(from == mesh->self) {
 				logger(mesh, MESHLINK_WARNING, "Got %s from %s for ourself which does not match existing entry",
 				       "ADD_EDGE", c->name);
-				send_add_edge(mesh, c, e);
+				send_add_edge(mesh, c, e, 0);
 				return true;
 			} else {
 				logger(mesh, MESHLINK_WARNING, "Got %s from %s which does not match existing entry",
@@ -148,7 +153,7 @@ bool add_edge_h(meshlink_handle_t *mesh, connection_t *c, const char *request) {
 		e = new_edge();
 		e->from = from;
 		e->to = to;
-		send_del_edge(mesh, c, e);
+		send_del_edge(mesh, c, e, mesh->contradicting_add_edge);
 		free_edge(e);
 		return true;
 	}
@@ -172,9 +177,9 @@ bool add_edge_h(meshlink_handle_t *mesh, connection_t *c, const char *request) {
 	return true;
 }
 
-bool send_del_edge(meshlink_handle_t *mesh, connection_t *c, const edge_t *e) {
-	return send_request(mesh, c, "%d %x %s %s", DEL_EDGE, rand(),
-	                    e->from->name, e->to->name);
+bool send_del_edge(meshlink_handle_t *mesh, connection_t *c, const edge_t *e, int contradictions) {
+	return send_request(mesh, c, "%d %x %s %s %d", DEL_EDGE, rand(),
+	                    e->from->name, e->to->name, contradictions);
 }
 
 bool del_edge_h(meshlink_handle_t *mesh, connection_t *c, const char *request) {
@@ -182,8 +187,9 @@ bool del_edge_h(meshlink_handle_t *mesh, connection_t *c, const char *request) {
 	char from_name[MAX_STRING_SIZE];
 	char to_name[MAX_STRING_SIZE];
 	node_t *from, *to;
+	int contradictions = 0;
 
-	if(sscanf(request, "%*d %*x "MAX_STRING" "MAX_STRING, from_name, to_name) != 2) {
+	if(sscanf(request, "%*d %*x "MAX_STRING" "MAX_STRING" %d", from_name, to_name, &contradictions) < 2) {
 		logger(mesh, MESHLINK_ERROR, "Got bad %s from %s", "DEL_EDGE", c->name);
 		return false;
 	}
@@ -216,6 +222,10 @@ bool del_edge_h(meshlink_handle_t *mesh, connection_t *c, const char *request) {
 		return true;
 	}
 
+	if(contradictions > 50) {
+		handle_duplicate_node(mesh, from);
+	}
+
 	/* Check if edge exists */
 
 	e = lookup_edge(from, to);
@@ -230,7 +240,7 @@ bool del_edge_h(meshlink_handle_t *mesh, connection_t *c, const char *request) {
 		logger(mesh, MESHLINK_WARNING, "Got %s from %s for ourself",
 		       "DEL_EDGE", c->name);
 		mesh->contradicting_del_edge++;
-		send_add_edge(mesh, c, e);    /* Send back a correction */
+		send_add_edge(mesh, c, e, mesh->contradicting_del_edge);    /* Send back a correction */
 		return true;
 	}
 
@@ -252,7 +262,7 @@ bool del_edge_h(meshlink_handle_t *mesh, connection_t *c, const char *request) {
 		e = lookup_edge(to, mesh->self);
 
 		if(e) {
-			send_del_edge(mesh, mesh->everyone, e);
+			send_del_edge(mesh, mesh->everyone, e, 0);
 			edge_del(mesh, e);
 		}
 	}
diff --git a/test/Makefile.am b/test/Makefile.am
index f364215b..0bc98d0c 100644
--- a/test/Makefile.am
+++ b/test/Makefile.am
@@ -1,13 +1,14 @@
 TESTS = \
 	basic.test \
 	basicpp.test \
-	trio.test \
 	channels.test \
 	channels-fork.test \
 	channels-cornercases.test \
+	duplicate.test \
 	import-export.test \
 	invite-join.test \
-	sign-verify.test
+	sign-verify.test \
+	trio.test
 
 dist_check_SCRIPTS = $(TESTS)
 
@@ -17,14 +18,15 @@ AM_LDFLAGS = $(PTHREAD_LIBS)
 check_PROGRAMS = \
 	basic \
 	basicpp \
-	trio \
 	channels \
 	channels-fork \
 	channels-cornercases \
+	duplicate \
+	echo-fork \
 	import-export \
 	invite-join \
 	sign-verify \
-	echo-fork
+	trio
 
 if INSTALL_TESTS
 bin_PROGRAMS = $(check_PROGRAMS)
@@ -36,9 +38,6 @@ basic_LDADD = ../src/libmeshlink.la
 basicpp_SOURCES = basicpp.cpp
 basicpp_LDADD = ../src/libmeshlink.la
 
-trio_SOURCES = trio.c
-trio_LDADD = ../src/libmeshlink.la
-
 channels_SOURCES = channels.c
 channels_LDADD = ../src/libmeshlink.la
 
@@ -48,6 +47,9 @@ channels_fork_LDADD = ../src/libmeshlink.la
 channels_cornercases_SOURCES = channels-cornercases.c utils.c utils.h
 channels_cornercases_LDADD = ../src/libmeshlink.la
 
+duplicate_SOURCES = duplicate.c
+duplicate_LDADD = ../src/libmeshlink.la
+
 echo_fork_SOURCES = echo-fork.c
 echo_fork_LDADD = ../src/libmeshlink.la
 
@@ -59,3 +61,6 @@ invite_join_LDADD = ../src/libmeshlink.la
 
 sign_verify_SOURCES = sign-verify.c
 sign_verify_LDADD = ../src/libmeshlink.la
+
+trio_SOURCES = trio.c
+trio_LDADD = ../src/libmeshlink.la