Adding ZeroMQ 4 support

This build verifies that everything compiles okay against ZeroMQ 4, and
adds the security features that come with it. Both the executor (mqexec)
and the event broker module now support curve authentication/encryption.
Authentication is done via an SSH authorized_keys-style file configured
at runtime.

Author: jbreams

dnxmq/mqbroker.c

@@ -91,7 +91,7 @@ void parse_sock_directive(json_t * arg, zmq_pollitem_t * pollable, int * noblock
 	char *type;
 #if ZMQ_VERSION_MAJOR == 2
 	int64_t hwm = 0, swap = 0, affinity = 0;
-#elif ZMQ_VERSION_MAJOR == 3
+#elif ZMQ_VERSION_MAJOR >= 3
 	int sndhwm = -1, rcvhwm = -1, maxmsgsize = 0, backlog = 0;
 	int64_t affinity = 0;
 	json_t * accept_filters = NULL;
@@ -103,7 +103,7 @@ void parse_sock_directive(json_t * arg, zmq_pollitem_t * pollable, int * noblock
 		"type", &type, "connect", &connect, "bind", &bind,
 		"subscribe", &subscribe, "hwm", &hwm, "swap", &swap,
 		"affinity", &affinity, "noblock", noblock) != 0)
-#elif ZMQ_VERSION_MAJOR == 3
+#elif ZMQ_VERSION_MAJOR >= 3
 	if(json_unpack(arg, "{s:s s?:o s?:o s?:o s?i s?i s?i s?b s?i s?o s?i}", 
 		"type", &type, "connect", &connect, "bind", &bind,
 		"subscribe", &subscribe, "sndhwm", &sndhwm, "rcvhwm", &rcvhwm,
@@ -153,7 +153,7 @@ void parse_sock_directive(json_t * arg, zmq_pollitem_t * pollable, int * noblock
 		zmq_setsockopt(sock, ZMQ_HWM, &hwm, sizeof(hwm));
 	if(swap > 0)
 		zmq_setsockopt(sock, ZMQ_SWAP, &swap, sizeof(swap));
-#elif ZMQ_VERSION_MAJOR == 3
+#elif ZMQ_VERSION_MAJOR >= 3
 	if(sndhwm > -1)
 		zmq_setsockopt(sock, ZMQ_SNDHWM, &sndhwm, sizeof(sndhwm));
 	if(rcvhwm > -1)
@@ -199,7 +199,7 @@ void do_forward(void * in, void *out, void *mon, int noblock, int monnoblock) {
 	int rc;
 #if ZMQ_VERSION_MAJOR == 2
 	int64_t rcvmore;
-#elif ZMQ_VERSION_MAJOR == 3
+#elif ZMQ_VERSION_MAJOR >= 3
 	int rcvmore;
 #endif
 	size_t size = sizeof(rcvmore);

dnxmq/mqexec.c

@@ -43,6 +43,9 @@ ev_io pullio;
 char * rootpath = NULL, *unprivpath = NULL;
 size_t rootpathlen = 0, unprivpathlen = 0;
 uid_t runas = 0;
+#if ZMQ_VERSION_MAJOR > 3
+char * curve_private = NULL, *curve_public = NULL, *curve_server = NULL;
+#endif
 
 struct child_job {
 	json_t * input;
@@ -512,7 +515,7 @@ void recv_job_cb(struct ev_loop * loop, ev_io * i, int event) {
 		zmq_msg_t inmsg;
 #if ZMQ_VERSION_MAJOR == 2
 		int64_t rcvmore = 0;
-#elif ZMQ_VERSION_MAJOR == 3
+#elif ZMQ_VERSION_MAJOR >= 3
 		int rcvmore = 0;
 #endif
 		size_t rms = sizeof(rcvmore);
@@ -549,13 +552,35 @@ void parse_sock_directive(void * socket, json_t * arg, int bind) {
 			zmq_bind(socket, json_string_value(arg));
 		else
 			zmq_connect(socket, json_string_value(arg));
+#if ZMQ_VERSION_MAJOR > 3
+		if(curve_private) {
+			zmq_setsockopt(socket, ZMQ_CURVE_SECRETKEY,
+				curve_private, strlen(curve_private));
+			zmq_setsockopt(socket, ZMQ_CURVE_PUBLICKEY,
+				curve_public, strlen(curve_public));
+			zmq_setsockopt(socket, ZMQ_CURVE_SERVERKEY,
+				curve_server, strlen(curve_server));
+		}
+#endif
 	} else if(json_is_object(arg)) {
 		char * addr = NULL;
 		json_t * subscribe = NULL;
 		if(json_unpack(arg, "{s:s s?:b s?:o}", 
 			"address", &addr, "bind", &bind, "subscribe",
 			&subscribe) != 0)
 			return;
+
+#if ZMQ_VERSION_MAJOR > 3
+		if(curve_private) {
+			zmq_setsockopt(socket, ZMQ_CURVE_SECRETKEY,
+				curve_private, strlen(curve_private));
+			zmq_setsockopt(socket, ZMQ_CURVE_PUBLICKEY,
+				curve_public, strlen(curve_public));
+			zmq_setsockopt(socket, ZMQ_CURVE_SERVERKEY,
+				curve_server, strlen(curve_server));
+		}
+#endif
+
 		if(bind)
 			zmq_bind(socket, addr);
 		else
@@ -610,6 +635,7 @@ int main(int argc, char ** argv) {
 	json_error_t config_err;
 	char ch, *configobj = "executor", *tmprootpath = NULL,
 		*tmpunprivpath = NULL, *tmpunprivuser = NULL;
+	json_error_t jsonerr;
 
 	while((ch = getopt(argc, argv, "vsdhc:")) != -1) {
 		switch(ch) {
@@ -657,15 +683,31 @@ int main(int argc, char ** argv) {
 		exit(1);
 	}
 
-	if(json_unpack(config, "{s:{s?:o s:o s?i s?b s?b s?:o s?o s?s s?s s?s}}",
+#if ZMQ_VERSION_MAJOR < 4
+	if(json_unpack_ex(config, &jsonerr, 0,
+		"{s:{s?:o s:o s?i s?b s?b s?:o s?o s?s s?s s?s}}",
 		configobj, "jobs", &jobs, "results", &results,
 		"iothreads", &iothreads, "verbose", &verbose,
 		"syslog", &usesyslog, "filter", &filter,
 		"publisher", &publisher, "rootpath", &tmprootpath,
 		"unprivpath", &tmpunprivpath, "unprivuser", &tmpunprivuser) != 0) {
-		logit(ERR, "Error getting config");
+		logit(ERR, "Error getting config %s", jsonerr.text);
 		exit(-1);
 	}
+#else
+	if(json_unpack_ex(config, &jsonerr, 0,
+		"{s:{s?:o s:o s?i s?b s?b s?:o s?o s?s s?s s?s s?{s:s s:s s:s}}}",
+		configobj, "jobs", &jobs, "results", &results,
+		"iothreads", &iothreads, "verbose", &verbose,
+		"syslog", &usesyslog, "filter", &filter,
+		"publisher", &publisher, "rootpath", &tmprootpath,
+		"unprivpath", &tmpunprivpath, "unprivuser", &tmpunprivuser,
+		"curve", "publickey", &curve_public, "privatekey", &curve_private,
+		"serverkey", &curve_server) != 0) {
+		logit(ERR, "Error getting config: %s", jsonerr.text);
+		exit(-1);
+	}
+#endif
 
 	parse_filter(filter,0);
 
@@ -697,6 +739,15 @@ int main(int argc, char ** argv) {
 		runas = pwdent->pw_uid;
 	}
 
+#if ZMQ_VERSION_MAJOR > 3
+	if(curve_public)
+		curve_public = strdup(curve_public);
+	if(curve_private)
+		curve_private = strdup(curve_private);
+	if(curve_server)
+		curve_server = strdup(curve_server);
+#endif
+
 	zmqctx = zmq_init(iothreads);
 	if(zmqctx == NULL)
 		exit(-1);