Implements message delivery.

Adds a minimum number of messages per mix.
Embeds the destination in the message.
Clients now send messages to themselves.

15 files changed, 156 insertions, 49 deletions

diff --git a/client/cmixclient.cpp b/client/cmixclient.cpp
index 861a6ac..5109b81 100644
--- a/client/cmixclient.cpp
+++ b/client/cmixclient.cpp
@@ -42,9 +42,10 @@ void CMixClient::initialize_connections() {
 			cmix_proto::ImAClient imaclient;
 			
 			char* id;
-			get_pub_key_hash(&cmix_ctx, &id);
+			size_t len;
+			get_pub_key_hash(&cmix_ctx, &id, &len);
 			
-			imaclient.set_id(id);
+			imaclient.set_id(id, len);
 			free(id);
 			
 			BOOST_LOG_TRIVIAL(trace) << "sending imaclient to node: " << i;
@@ -59,6 +60,11 @@ void CMixClient::initialize_connections() {
 	}
 }
 
+void CMixClient::handle_payload(cmix_proto::Payload pl)
+{
+	std::cout << "Received message: " << pl.payload() << std::endl;
+}
+
 void CMixClient::handle_key_exchange(size_t node_id, cmix_proto::KeyExchange const& ke)
 {
 	key_exchange_initiator(
@@ -71,9 +77,18 @@ void CMixClient::handle_key_exchange(size_t node_id, cmix_proto::KeyExchange con
 	
 	if(std::all_of(shared_values.begin(), shared_values.end(), [](GroupElement const& value){return value != nullptr;})) {
 		size_t len = get_group_element_array_size(&cmix_ctx);
+		
 		std::vector<char> vec(len, '\0');
+		
+		char* id;
+		size_t id_len;
+		
+		get_pub_key_hash(&cmix_ctx, &id, &id_len);
 		std::string s = "abcdefghijklmnopqrstuvwxyz";
-		std::copy_n(s.begin(), s.size(), vec.begin() + 1);
+		
+		create_message(&cmix_ctx, vec.data(), id, id_len, s.data(), s.size());
+		
+		free(id);
 		
 		cmix_proto::UserMessage message;
 		message.mutable_m()->resize(len);
@@ -83,12 +98,12 @@ void CMixClient::handle_key_exchange(size_t node_id, cmix_proto::KeyExchange con
 		BOOST_LOG_TRIVIAL(trace) << "sending UserMessage: " << message.ShortDebugString();
 		network_connections.at(0).async_send(message);
 		
-//		size_t last_node_id = network_details.node_details.size() - 1;
-//		network_connections.at(last_node_id).async_receive(
-//			[this, last_node_id](cmix_proto::CMixMessage const& message) {
-//				handle_message(last_node_id, message);
-//			}
-//		);
+		size_t last_node_id = network_details.node_details.size() - 1;
+		network_connections.at(last_node_id).async_receive(
+			[this, last_node_id](cmix_proto::CMixMessage const& message) {
+				handle_message(last_node_id, message);
+			}
+		);
 	}
 }
 
@@ -114,6 +129,11 @@ void CMixClient::handle_message(size_t node_id, cmix_proto::CMixMessage message)
 			key_exchange(node_id);
 			return;
  		}
+ 		case cmix_proto::CMixMessage::ContentsCase::kPayload: {
+			BOOST_LOG_TRIVIAL(trace) << "handling Payload: ";
+			handle_payload(message.payload());
+			return;
+ 		}
 		default: {
 			BOOST_LOG_TRIVIAL(error) << "Received unknown message";
 		}
diff --git a/client/cmixclient.hpp b/client/cmixclient.hpp
index 30c096d..6291bd1 100644
--- a/client/cmixclient.hpp
+++ b/client/cmixclient.hpp
@@ -61,9 +61,10 @@ class CMixClient {
 
 	void initialize_connections();
 
+	void handle_payload(cmix_proto::Payload pl);
 	void handle_key_exchange(size_t node_id, cmix_proto::KeyExchange const& ke);
-
 	void handle_message(size_t node_id, cmix_proto::CMixMessage message);
+	
 
 public:
 	/*!
diff --git a/libcmix-common/cmixprotofunctor.hpp b/libcmix-common/cmixprotofunctor.hpp
index 0055593..57536e5 100644
--- a/libcmix-common/cmixprotofunctor.hpp
+++ b/libcmix-common/cmixprotofunctor.hpp
@@ -67,6 +67,7 @@ struct CMixProtoFunctor {
 		PrePost, prepost,
 		RealPre, realpre,
 		RealMix, realmix,
+		Payload, payload
 	)
 	#undef MESSAGE_SETTER_DEFS
 	#undef MESSAGE_SETTER_DEF_ITERATION
diff --git a/libcmix-crypto/api.h b/libcmix-crypto/api.h
index caf97b6..0fa4916 100644
--- a/libcmix-crypto/api.h
+++ b/libcmix-crypto/api.h
@@ -66,6 +66,11 @@ typedef GroupElement(*ArrayToElement)(char const*, size_t size, bool);
 typedef void(*PubKeyHashGetter)(char** buffer, size_t* len, GroupElement const pub);
 
 /*!
+ *
+ */
+typedef size_t(*PubKeyHashLengthGetter)();
+
+/*!
  * \brief PublicShareAdder typedef
  */
 typedef void(*PublicShareAdder)(GroupElement*, char const*, size_t, GroupElement);
@@ -124,8 +129,9 @@ struct Api {
 	KeyPairDeleter free_keypair; ///< Pointer to keypair deletor function
 	ElementToArray element_to_array; ///< Get the array representation of a public key
 	BufferDeleter free_buffer; ///< frees library allocated buffers.
-	ArrayToElement array_to_element; ///< The the GroupElement representation of this array;
+	ArrayToElement array_to_element; ///< The the GroupElement representation of this array.
 	PubKeyHashGetter get_pub_key_hash; ///< Get the hash of the public key.
+	PubKeyHashLengthGetter get_pub_key_hash_length; ///< Get the length of the pubkey hash.
 	GroupElementGetter get_group_element; ///< get group element
 	GroupElementArraySizeGetter get_group_element_array_size; ///< Return the size required to store a groupelement in an array;
 	GroupElementDeleter free_group_element; ///< frees a base type of the cryptolibrary.
diff --git a/libcmix-crypto/elgamal/elgamal.c b/libcmix-crypto/elgamal/elgamal.c
index 8f26a3c..399d952 100644
--- a/libcmix-crypto/elgamal/elgamal.c
+++ b/libcmix-crypto/elgamal/elgamal.c
@@ -11,6 +11,7 @@ struct Api get_elgamal_implementation()
 		.free_buffer = elgamal_free_buffer,
 		.array_to_element = elgamal_array_to_element,
 		.get_pub_key_hash = elgamal_get_pub_key_hash,
+		.get_pub_key_hash_length = elgamal_get_pub_key_hash_length,
 		.get_group_element = elgamal_get_group_element,
 		.get_group_element_array_size = elgamal_get_group_element_array_size,
 		.free_group_element = elgamal_delete_group_element,
diff --git a/libcmix-crypto/elgamal/elgamal.h b/libcmix-crypto/elgamal/elgamal.h
index f937899..1fe4447 100644
--- a/libcmix-crypto/elgamal/elgamal.h
+++ b/libcmix-crypto/elgamal/elgamal.h
@@ -24,6 +24,8 @@ extern ArrayToElement elgamal_array_to_element;
 
 extern PubKeyHashGetter elgamal_get_pub_key_hash;
 
+extern PubKeyHashLengthGetter elgamal_get_pub_key_hash_length;
+
 extern GroupElementGetter elgamal_get_group_element;
 
 extern GroupElementDeleter elgamal_delete_group_element;
diff --git a/libcmix-crypto/elgamal/gcrypt/gcrypt_elgamal.c b/libcmix-crypto/elgamal/gcrypt/gcrypt_elgamal.c
index c9db28c..28c61c3 100644
--- a/libcmix-crypto/elgamal/gcrypt/gcrypt_elgamal.c
+++ b/libcmix-crypto/elgamal/gcrypt/gcrypt_elgamal.c
@@ -158,6 +158,10 @@ void gcrypt_elgamal_get_pub_key_hash(char** buffer, size_t* len, GroupElement co
 	gcry_sexp_release(pubkey);
 }
 
+size_t gcrypt_elgamal_get_pub_key_hash_length() {
+	return 20u;
+}
+
 void* gcrypt_elgamal_get_group_element(bool secure) {
 	size_t parse_error_offset;
 	gcry_error_t error;
@@ -337,6 +341,7 @@ ElementToArray elgamal_element_to_array = &gcrypt_elgamal_element_to_array;
 BufferDeleter elgamal_free_buffer = &gcrypt_elgamal_free_buffer;
 ArrayToElement elgamal_array_to_element = &gcrypt_elgamal_array_to_element;
 PubKeyHashGetter elgamal_get_pub_key_hash = &gcrypt_elgamal_get_pub_key_hash;
+PubKeyHashLengthGetter elgamal_get_pub_key_hash_length = &gcrypt_elgamal_get_pub_key_hash_length;
 GroupElementGetter elgamal_get_group_element = &gcrypt_elgamal_get_group_element;
 GroupElementDeleter elgamal_delete_group_element = &gcrypt_elgamal_delete_group_element;
 KeyExchangeValueGetter elgamal_get_key_exchange_value = &gcrypt_elgamal_get_key_exchange_value;
diff --git a/libcmix-protobuf/cmix.proto b/libcmix-protobuf/cmix.proto
index 8bb1d42..d0e3da2 100644
--- a/libcmix-protobuf/cmix.proto
+++ b/libcmix-protobuf/cmix.proto
@@ -13,7 +13,7 @@ message ImANode {
 }
 
 message ImAClient {
-	required string id = 1;
+	required bytes id = 1;
 }
 
 message Bye {
@@ -57,6 +57,10 @@ message RealMix {
 	repeated bytes m = 2;
 }
 
+message Payload {
+	required bytes payload = 1;
+}
+
 message CMixMessage {
 	oneof contents {
 		Initialization initialization = 1;
@@ -72,5 +76,6 @@ message CMixMessage {
 		PrePost prepost = 11;
 		RealPre realpre = 12;
 		RealMix realmix = 13;
+		Payload payload = 14;
 	}
 }
diff --git a/libcmix/cmix.c b/libcmix/cmix.c
index 00b080f..2adca78 100644
--- a/libcmix/cmix.c
+++ b/libcmix/cmix.c
@@ -4,6 +4,7 @@
 #include <string.h>
 #include <stdlib.h>
 #include <stdio.h>
+#include <math.h>
 
 enum cmix_error calculate_shared_key_part(struct Bignum* result, struct Bignum partial_shared, struct Bignum my_share, struct Bignum mod)
 {
@@ -380,20 +381,38 @@ enum cmix_error remove_r_and_s(struct CMixContext const* ctx, char* out_buffer,
 	ctx->api.free_group_element(mult);
 }
 
-enum cmix_error get_pub_key_hash(struct CMixContext const* ctx, char** buffer)
+enum cmix_error get_pub_key_hash(struct CMixContext const* ctx, char** buffer, size_t* len) {
+	ctx->api.get_pub_key_hash(buffer, len, ctx->keypair.pub);
+	
+	return no_error;
+}
+
+size_t get_pub_key_hash_length(struct CMixContext const* ctx)
 {
-	char* buffer2;
-	size_t len;
-	ctx->api.get_pub_key_hash(&buffer2, &len, ctx->keypair.pub);
+	return ctx->api.get_pub_key_hash_length();
+}
+
+enum cmix_error create_message(struct CMixContext const* ctx, char* out_buffer, const char* dest, size_t dest_len, const char* payload, size_t payload_len)
+{
+	size_t el_arr_len = get_group_element_array_size(ctx);
 	
-	size_t hex_len = len*2+1;
-	*buffer = (char*) calloc(hex_len, sizeof(char));
+	memcpy(&out_buffer[1], dest, dest_len);
 	
-	for(size_t i = 0; i < len; i++) {
-		sprintf(&(*buffer)[i*2],"%hhX",buffer2[i]);
-	}
+	size_t message_limit = el_arr_len - 1 -dest_len;
 	
-	ctx->api.free_buffer(buffer2);
+	memcpy(&out_buffer[1] + dest_len, payload, (message_limit < payload_len ? message_limit : payload_len));
+}
+
+enum cmix_error split_message(struct CMixContext const* ctx, char** dest_buffer, size_t* dest_len, char** payload_buffer, size_t* payload_len, char const* message)
+{
+	size_t el_arr_len = get_group_element_array_size(ctx);
+	*dest_len = get_pub_key_hash_length(ctx);
 	
-	return no_error;
+	*dest_buffer = (char*) calloc(*dest_len, sizeof(char));
+	memcpy(*dest_buffer, &message[1], *dest_len);
+	
+	*payload_len = el_arr_len - 1 - *dest_len;
+	*payload_buffer = (char*) calloc(*payload_len, sizeof(char));
+	memcpy(*payload_buffer, &message[1 + *dest_len], *payload_len);
 }
+
diff --git a/libcmix/cmix.h b/libcmix/cmix.h
index 8e6f598..c3eeb3a 100644
--- a/libcmix/cmix.h
+++ b/libcmix/cmix.h
@@ -69,7 +69,9 @@ enum cmix_error get_public_key(struct CMixContext const* ctx, char* buffer);
 
 enum cmix_error add_public_share(struct CMixContext const* ctx, char* buffer, char const* share);
 
-enum cmix_error get_pub_key_hash(struct CMixContext const* ctx, char** buffer);
+enum cmix_error get_pub_key_hash(struct CMixContext const* ctx, char** buffer, size_t* len);
+
+size_t get_pub_key_hash_length(struct CMixContext const* ctx);
 
 enum cmix_error start_mix(struct CMixContext* ctx, size_t nr_participants);
 
@@ -111,6 +113,10 @@ enum cmix_error swap_k_for_r(struct CMixContext const* ctx, char** out_buffer, c
 
 enum cmix_error remove_r_and_s(struct CMixContext const* ctx, char* out_buffer, char const* message, size_t index);
 
+enum cmix_error create_message(struct CMixContext const* ctx, char* out_buffer, char const* dest, size_t dest_len, char const* payload, size_t payload_len);
+
+enum cmix_error split_message(struct CMixContext const* ctx, char** dest_buffer, size_t* dest_len, char** payload_buffer, size_t* payload_len, char const* message);
+
 #ifdef __cplusplus
 } // extern "C"
 #endif
diff --git a/node/main.cpp b/node/main.cpp
index acfdafe..02838f1 100644
--- a/node/main.cpp
+++ b/node/main.cpp
@@ -22,6 +22,7 @@ int main(int argc, char* argv[]) {
 		("next_node,n", po::value<std::string>(), "The address of the next node in the network.")
 		("first,f", "This is the first node and will be the communication point for the clients.")
 		("last,l", "this is the last node and will be able to send the original message out of the network.")
+		("minimum_nr_messages,m", po::value<unsigned int>(), "minimum number of messages that needs to accumulate before starting a mix.")
 		("cert,c", po::value<std::string>(), "The cert file to use (in pem format).")
 		("key,k", po::value<std::string>(), "The key file (in pem format).")
 		("dhparam,d", po::value<std::string>(), "The dhparam file.")
@@ -93,6 +94,15 @@ int main(int argc, char* argv[]) {
 	BOOST_LOG_TRIVIAL(info) << "Started node";
 
 	bool is_first = bool(vm.count("first"));
+	unsigned int minimum_nr_messages = 0;
+	if(is_first) {
+		if(vm.count("minimum_nr_messages")) {
+			minimum_nr_messages = vm["minimum_nr_messages"].as<unsigned int>();
+		} else {
+			std::cerr << "Minimum nr of messages is a required parameter for the first node" << std::endl;
+			return -1;
+		}
+	}
 	bool is_last = bool(vm.count("last"));
 	std::string next_node;
 	if(vm.count("next_node")) {
@@ -114,7 +124,7 @@ int main(int argc, char* argv[]) {
 	
 	Uri uri = parse_uri(next_node);
 	
-	NodeNetworkSettings nsettings{is_first, is_last, uri.host, uri.port, certdir};
+	NodeNetworkSettings nsettings{is_first, is_last, uri.host, uri.port, certdir, minimum_nr_messages};
 	
 	Node node(lsettings, nsettings);
 	node.run();
diff --git a/node/node.cpp b/node/node.cpp
index 2310bc1..84ea179 100644
--- a/node/node.cpp
+++ b/node/node.cpp
@@ -156,8 +156,14 @@ void Node::handle_message(Purgatory::iterator handle, cmix_proto::CMixMessage me
 }
 
 void Node::start_precomputation() {
-	BOOST_LOG_TRIVIAL(trace) << "Starting precomputation for " << clients.size() << " clients.";
+	BOOST_LOG_TRIVIAL(trace) << "Starting precomputation for " << messages.size() << " clients.";
 	index_map.clear();
+	
+	if(messages.size() < network_settings.minimum_nr_messages) {
+		start_timer_delayed_mix();
+		return;
+	}
+	
 	if(start_mix(&cmix_ctx, messages.size()) != no_error) {
 		exit(-1);
 	}
@@ -278,3 +284,17 @@ bool Node::send_bye(bool got_bye)
 		return false;
 	}
 }
+
+void Node::start_timer_delayed_mix()
+{
+	timer.expires_from_now(boost::posix_time::millisec(500));
+	timer.async_wait([this](boost::system::error_code const& ec) {
+		if(ec == boost::system::errc::operation_canceled) {
+			return;
+		}
+		if(ec) {
+			throw std::runtime_error(ec.message());
+		}
+		start_precomputation();
+	});
+}
diff --git a/node/node.hpp b/node/node.hpp
index db3c66f..5365c9f 100644
--- a/node/node.hpp
+++ b/node/node.hpp
@@ -40,6 +40,7 @@ struct NodeNetworkSettings {
 	std::string next_host; ///< Next nodes host.
 	std::string next_port; ///< Next nodes port.
 	std::string certdir; ///< Directory containing trusted certificate authorities.
+	unsigned int minimum_nr_messages; ///< The minimum number of available messages before starting to run a mix;
 };
 
 /*!
@@ -85,6 +86,7 @@ class Node
 	void start_realtime_phase();
 	void shutdown();
 	bool send_bye(bool got_bye);
+	void start_timer_delayed_mix();
 	
 	void handle_node_initialization(cmix_proto::Initialization const& init);
 	void handle_node_secretkey(cmix_proto::SecretKey const& secret);
diff --git a/node/node_node.cpp b/node/node_node.cpp
index f68b496..00968a0 100644
--- a/node/node_node.cpp
+++ b/node/node_node.cpp
@@ -205,10 +205,7 @@ void Node::handle_node_secretkey(cmix_proto::SecretKey const& secret)
 	std::string share = secret.secret_key();
 	
 	if(network_settings.is_first) {
-		timer.expires_from_now(boost::posix_time::seconds(4));
-		timer.async_wait([this](boost::system::error_code const& ec) {
-			start_precomputation();
-		});
+		start_timer_delayed_mix();
 	} else {
 		set_network_key(&cmix_ctx, secret.secret_key().data(), secret.secret_key().size());
 		next_node.async_send(secret);
@@ -245,27 +242,16 @@ void Node::handle_node_prepost(cmix_proto::PrePost const& prepost) {
 }
 
 void Node::handle_node_realpre(cmix_proto::RealPre const& realpre) {
-	auto final = [this](){
-		deinitialize(&cmix_ctx);
-		shutdown();
-	};
-	
 	if(network_settings.is_first) {
 		cmix_proto::RealMix n_realmix = fill_realtime_mix_message(cmix_ctx, realpre.m());
-		next_node.async_send(n_realmix, final);
+		next_node.async_send(n_realmix);
 	} else {
 		cmix_proto::RealPre n_realpre = fill_realtime_pre_message(cmix_ctx, realpre.h(), realpre.m(), data);
 		next_node.async_send(n_realpre);
 	}
 }
 
-void Node::handle_node_realmix(cmix_proto::RealMix const& realmix) {
-	auto final = [this](){
-		deinitialize(&cmix_ctx);
-		shutdown();
-		send_bye(false);
-	};
-	
+void Node::handle_node_realmix(cmix_proto::RealMix const& realmix) {	
 	if(network_settings.is_last) {
 		BOOST_LOG_TRIVIAL(trace) << "Doing the last step:";
 		
@@ -277,18 +263,31 @@ void Node::handle_node_realmix(cmix_proto::RealMix const& realmix) {
 		
 		for(int i = 0; i < n_realmix.m_size(); i++) {
 			remove_r_and_s(&cmix_ctx, &str[0], n_realmix.m(i).data(), i);
-			
 			{
 				std::stringstream ss;
 				for(auto&& c : str) {
 					ss << "\\" << std::setw(3) << std::setfill('0') << std::oct << (unsigned int) c;
 				}
 				BOOST_LOG_TRIVIAL(trace) << ss.str();
-			} 
+			}
+			
+			char* dest;
+			size_t dest_len;
+			
+			char* payload;
+			size_t payload_len;
+			
+			split_message(&cmix_ctx, &dest, &dest_len, &payload, &payload_len, str.data());
+			
+			std::string dest_s = std::string(dest, dest_len);
+			
+			cmix_proto::Payload pay;
+			pay.set_payload(payload, payload_len);
+			clients.at(dest_s).async_send(pay);
+			
+			free(dest);
+			free(payload);
 		}
-		
-		final();
-		
 	} else {
 		cmix_proto::RealMix n_realmix = fill_realtime_mix_message(cmix_ctx, realmix.m());
 		next_node.async_send(n_realmix);
diff --git a/run.sh b/run.sh
index 2d6b8ac..7240167 100755
--- a/run.sh
+++ b/run.sh
@@ -3,7 +3,7 @@
 build_dir=/home/dennis/projects/cmix/build-cmix-Desktop-Default
 
 tmux new-session -s cmix -d
-tmux send-keys -t cmix:0 "cd ${build_dir} && LSAN_OPTIONS=report_objects=1 node/node -f -n node2.local:9201 -c ../certs/cert1.pem -k ../certs/key1.pem -d ../certs/dh.pem --certdir ../certs/" Enter
+tmux send-keys -t cmix:0 "cd ${build_dir} && LSAN_OPTIONS=report_objects=1 node/node -f -m 5 -n node2.local:9201 -c ../certs/cert1.pem -k ../certs/key1.pem -d ../certs/dh.pem --certdir ../certs/" Enter
 #tmux send-keys -t cmix:0 "r" Enter
 
 tmux new-window -t cmix:1
@@ -16,5 +16,15 @@ tmux send-keys -t cmix:2 "cd ${build_dir} && LSAN_OPTIONS=report_objects=1 clien
 tmux new-window -t cmix:3
 tmux send-keys -t cmix:3 "cd ${build_dir} && LSAN_OPTIONS=report_objects=1 client/client --certdir ../certs -n node1.local:9200 node2.local:9201" Enter
 
+tmux new-window -t cmix:4
+tmux send-keys -t cmix:4 "cd ${build_dir} && LSAN_OPTIONS=report_objects=1 client/client --certdir ../certs -n node1.local:9200 node2.local:9201" Enter
+
+tmux new-window -t cmix:5
+tmux send-keys -t cmix:5 "cd ${build_dir} && LSAN_OPTIONS=report_objects=1 client/client --certdir ../certs -n node1.local:9200 node2.local:9201" Enter
+
+tmux new-window -t cmix:6
+tmux send-keys -t cmix:6 "cd ${build_dir} && LSAN_OPTIONS=report_objects=1 client/client --certdir ../certs -n node1.local:9200 node2.local:9201" Enter
+
+
 tmux attach
 tmux kill-session -t cmix