Netconf service code, interacts with our MySQL database.

3 files changed, 232 insertions, 13 deletions

diff --git a/netconf-plugin/Makefile b/netconf-plugin/Makefile
index e69de29b..bd6b052c 100644
--- a/netconf-plugin/Makefile
+++ b/netconf-plugin/Makefile
@@ -0,0 +1,6 @@
+all:
+	gcc -O6 -c ../ext/lz4/lz4hc.c ../ext/lz4/lz4.c
+	g++ -DZT_OSNAME="linux" -DZT_ARCH="x86_64" -I/usr/include/mysql -I../ext/bin/libcrypto/include -O -o netconf.service netconf.cpp ../node/Utils.cpp ../node/Identity.cpp ../node/EllipticCurveKeyPair.cpp ../node/Salsa20.cpp ../node/HMAC.cpp lz4.o lz4hc.o ../ext/bin/libcrypto/linux-x86_64/libcrypto.a -lmysqlpp
+
+clean:
+	rm -f *.o netconf.service
diff --git a/netconf-plugin/netconf.cpp b/netconf-plugin/netconf.cpp
index 57d3653b..a92ff8a0 100644
--- a/netconf-plugin/netconf.cpp
+++ b/netconf-plugin/netconf.cpp
@@ -52,8 +52,12 @@
 
 #include <stdio.h>
 #include <stdlib.h>
-#include <unistd.h>
 #include <string.h>
+#include <stdint.h>
+#include <unistd.h>
+#include <sys/stat.h>
+#include <sys/types.h>
+#include <arpa/inet.h>
 
 #include <iostream>
 #include <string>
@@ -62,32 +66,35 @@
 #include <vector>
 #include <algorithm>
 
-#include <mysql++.h>
+#include <mysql++/mysql++.h>
 
 #include "../node/Dictionary.hpp"
+#include "../node/Identity.hpp"
+#include "../node/Utils.hpp"
 
 using namespace ZeroTier;
 using namespace mysqlpp;
 
 static Connection *dbCon = (Connection *)0;
+static char mysqlHost[64],mysqlPort[64],mysqlDatabase[64],mysqlUser[64],mysqlPassword[64];
 
 static void connectOrReconnect()
 {
-	if (dbCon)
-		delete dbCon;
-	dbCon = new Connection(mysqlDatabase,mysqlHost,mysqlUser,mysqlPassword,(unsigned int)strtol(mysqlPort,(char **)0,10));
-	if (dbCon->connected())
-		break;
-	else {
-		fprintf(stderr,"Unable to connect to database server.\n");
-		usleep(1000);
+	for(;;) {
+		if (dbCon)
+			delete dbCon;
+		dbCon = new Connection(mysqlDatabase,mysqlHost,mysqlUser,mysqlPassword,(unsigned int)strtol(mysqlPort,(char **)0,10));
+		if (dbCon->connected())
+			break;
+		else {
+			fprintf(stderr,"Unable to connect to database server.\n");
+			usleep(1000);
+		}
 	}
 }
 
 int main(int argc,char **argv)
 {
-	char mysqlHost[64],mysqlPort[64],mysqlDatabase[64],mysqlUser[64],mysqlPassword[64];
-
 	{
 		char *ee = getenv("ZT_NETCONF_MYSQL_HOST");
 		if (!ee) {
@@ -96,7 +103,7 @@ int main(int argc,char **argv)
 		}
 		strcpy(mysqlHost,ee);
 		ee = getenv("ZT_NETCONF_MYSQL_PORT");
-		if (ee == null)
+		if (!ee)
 			strcpy(mysqlPort,"3306");
 		else strcpy(mysqlPort,ee);
 		ee = getenv("ZT_NETCONF_MYSQL_DATABASE");
@@ -119,9 +126,172 @@ int main(int argc,char **argv)
 		strcpy(mysqlPassword,ee);
 	}
 
+	char buf[4096];
+	std::string dictBuf;
+
 	connectOrReconnect();
 	for(;;) {
+		if (read(STDIN_FILENO,buf,4) != 4) {
+			fprintf(stderr,"Error reading frame size from stdin\n");
+			return -1;
+		}
+		unsigned int fsize = (unsigned int)ntohl(*((const uint32_t *)buf));
+		while (dictBuf.length() < fsize) {
+			int n = (int)read(STDIN_FILENO,buf,std::min((int)sizeof(buf),(int)(fsize - dictBuf.length())));
+			for(int i=0;i<n;++i)
+				dictBuf.push_back(buf[i]);
+		}
+		Dictionary msg(dictBuf);
+		dictBuf = "";
+
 		if (!dbCon->connected())
 			connectOrReconnect();
+
+		try {
+			const std::string &command = msg.get("command");
+			if (command == "config") { // NETWORK_CONFIG_REQUEST packet
+				Identity peerIdentity(msg.get("peerIdentity"));
+				uint64_t nwid = strtoull(msg.get("nwid").c_str(),(char **)0,16);
+				Dictionary meta;
+				if (msg.contains("meta"))
+					meta.fromString(msg.get("meta"));
+
+				// Do quick signature check / sanity check
+				if (!peerIdentity.locallyValidate(false)) {
+					fprintf(stderr,"identity failed signature check: %s",peerIdentity.toString(false).c_str());
+					continue;
+				}
+
+				// Save identity if unknown
+				{
+					Query q = dbCon->query();
+					q << "SELECT identity,identityValidated FROM Node WHERE id = " << peerIdentity.address().toInt();
+					StoreQueryResult rs = q.store();
+					if (rs.num_rows() > 0) {
+						if (rs[0]["identity"] != peerIdentity.toString(false)) {
+							// TODO: handle collisions...
+							continue;
+						} else if ((int)rs[0]["identityValidated"] == 0) {
+							// TODO: launch background validation
+						}
+					} else {
+						q = dbCon->query();
+						uint64_t now = Utils::now();
+						q << "INSERT INTO Node (id,creationTime,lastSeen,identity) VALUES (" << peerIdentity.address().toInt() << "," << now << "," << now << "," << peerIdentity.toString(false) << ")";
+						if (!q.exec()) {
+							fprintf(stderr,"Error inserting Node row for peer %s, aborting netconf request",peerIdentity.address().toString().c_str());
+							continue;
+						}
+						// TODO: launch background validation
+					}
+				}
+
+				bool isOpen = false;
+				{
+					Query q = dbCon->query();
+					q << "SELECT isOpen FROM Network WHERE id = " << nwid;
+					StoreQueryResult rs = q.store();
+					if (rs.num_rows() > 0)
+						isOpen = ((int)rs[0]["isOpen"] > 0);
+				}
+
+				Dictionary netconf;
+
+				netconf["peer"] = peerIdentity.address().toString();
+				sprintf(buf,"%.16llx",(unsigned long long)nwid);
+				netconf["nwid"] = buf;
+				netconf["isOpen"] = (isOpen ? "1" : "0");
+
+				if (!isOpen) {
+					// TODO: handle closed networks, look up private membership,
+					// generate signed cert.
+				}
+
+				std::string ipv4Static,ipv6Static;
+
+				{
+					// Check for IPv4 static assignments
+					Query q = dbCon->query();
+					q << "SELECT INET_NTOA(ip) AS ip,netmaskBits FROM IPv4Static WHERE Node_id = " << peerIdentity.address().toInt() << " AND Network_id = " << nwid;
+					StoreQueryResult rs = q.store();
+					if (rs.num_rows() > 0) {
+						for(int i=0;i<rs.num_rows();++i) {
+							if (ipv4Static.length())
+								ipv4Static.push_back(',');
+							ipv4Static.append(rs[i]["ip"].c_str());
+							ipv4Static.push_back('/');
+							ipv4Static.append(rs[i]["netmaskBits"].c_str());
+						}
+					}
+
+					// Try to auto-assign if there's any auto-assign networks with space
+					// available.
+					if (!ipv4Static.length()) {
+						unsigned char addressBytes[5];
+						peerIdentity.address().copyTo(addressBytes,5);
+
+						q = dbCon->query();
+						q << "SELECT ipNet,netmaskBits FROM IPv4AutoAssign WHERE Network_id = " << nwid;
+						rs = q.store();
+						if (rs.num_rows() > 0) {
+							for(int aaRow=0;aaRow<rs.num_rows();++aaRow) {
+								uint32_t ipNet = (uint32_t)((unsigned long)rs[aaRow]["ipNet"]);
+								unsigned int netmaskBits = (unsigned int)rs[aaRow]["netmaskBits"];
+
+								uint32_t tryIp = (((uint32_t)addressBytes[1]) << 24) |
+								                 (((uint32_t)addressBytes[2]) << 16) |
+								                 (((uint32_t)addressBytes[3]) << 8) |
+								                 ((((uint32_t)addressBytes[4]) % 254) + 1);
+								tryIp &= (0xffffffff >> netmaskBits);
+								tryIp |= ipNet;
+
+								for(int k=0;k<100000;++k) {
+									Query q2 = dbCon->query();
+									q2 << "INSERT INTO IPv4Static (Network_id,Node_id,ip,netmaskBits) VALUES (" << nwid << "," << peerIdentity.address().toInt() << "," << tryIp << "," << netmaskBits << ")";
+									if (q2.exec()) {
+										sprintf(buf,"%u.%u.%u.%u",(unsigned int)((tryIp >> 24) & 0xff),(unsigned int)((tryIp >> 16) & 0xff),(unsigned int)((tryIp >> 8) & 0xff),(unsigned int)(tryIp & 0xff));
+										if (ipv4Static.length())
+											ipv4Static.push_back(',');
+										ipv4Static.append(buf);
+										ipv4Static.push_back('/');
+										sprintf(buf,"%u",netmaskBits);
+										ipv4Static.append(buf);
+										break;
+									} else { // insert will fail if IP is in use due to uniqueness constraints in DB
+										++tryIp;
+										if ((tryIp & 0xff) == 0)
+											tryIp |= 1;
+										tryIp &= (0xffffffff >> netmaskBits);
+										tryIp |= ipNet;
+									}
+								}
+
+								if (ipv4Static.length())
+									break;
+							}
+						}
+					}
+				}
+
+				if (ipv4Static.length())
+					netconf["ipv4Static"] = ipv4Static;
+				if (ipv6Static.length())
+					netconf["ipv6Static"] = ipv6Static;
+
+				Dictionary resp;
+				resp["peer"] = peerIdentity.address().toString();
+				resp["nwid"] = msg.get("nwid");
+				resp["requestId"] = msg.get("requestId");
+				resp["netconf"] = netconf.toString();
+				std::string respm = resp.toString();
+				uint32_t respml = (uint32_t)htonl((uint32_t)respm.length());
+				write(STDOUT_FILENO,&respml,4);
+				write(STDOUT_FILENO,respm.data(),respm.length());
+			}
+		} catch (std::exception &exc) {
+			fprintf(stderr,"unexpected exception handling message: %s",exc.what());
+		} catch ( ... ) {
+			fprintf(stderr,"unexpected exception handling message: unknown exception");
+		}
 	}
 }
diff --git a/node/Service.cpp b/node/Service.cpp
index e0d06792..88a6d15c 100644
--- a/node/Service.cpp
+++ b/node/Service.cpp
@@ -109,9 +109,14 @@ bool Service::send(const Dictionary &msg)
 void Service::main()
 	throw()
 {
+	char buf[4096];
 	fd_set readfds,writefds,exceptfds;
 	struct timeval tv;
 
+	std::string stderrBuf;
+	std::string stdoutBuf;
+	unsigned int stdoutExpecting = 0;
+
 	while (_run) {
 		if (_pid <= 0) {
 			LOG("launching service %s...",_name.c_str());
@@ -133,10 +138,15 @@ void Service::main()
 				_childStdin = in[1];
 				_childStdout = out[0];
 				_childStderr = err[0];
+				fcntl(_childStdout,F_SETFL,O_NONBLOCK);
+				fcntl(_childStderr,F_SETFL,O_NONBLOCK);
 			} else {
 				dup2(in[0],STDIN_FILENO);
 				dup2(out[1],STDOUT_FILENO);
 				dup2(err[1],STDERR_FILENO);
+				close(in[1]);
+				close(out[0]);
+				close(err[0]);
 				execl(_path.c_str(),_path.c_str(),_r->homePath.c_str(),(const char *)0);
 				exit(-1);
 			}
@@ -179,9 +189,42 @@ void Service::main()
 		}
 
 		if ((_childStderr > 0)&&(FD_ISSET(_childStderr,&readfds))) {
+			int n = (int)read(_childStderr,buf,sizeof(buf));
+			for(int i=0;i<n;++i) {
+				if ((buf[i] == '\r')||(buf[i] == '\n')) {
+					stderrBuf = Utils::trim(stderrBuf);
+					if (stderrBuf.length())
+						LOG("service %s: %s",_name.c_str(),stderrBuf.c_str());
+					stderrBuf = "";
+				} else stderrBuf.push_back(buf[i]);
+			}
 		}
 
 		if ((_childStdout > 0)&&(FD_ISSET(_childStdout,&readfds))) {
+			int n = (int)read(_childStdout,buf,sizeof(buf));
+			for(int i=0;i<n;++i) {
+				stdoutBuf.push_back(buf[i]);
+				if (stdoutExpecting) {
+					if (stdoutBuf.length() == stdoutExpecting) {
+						try {
+							_handler(_arg,*this,Dictionary(stdoutBuf));
+						} catch ( ... ) {
+							LOG("unexpected exception handling message from service %s",_name.c_str());
+						}
+						stdoutBuf = "";
+						stdoutExpecting = 0;
+					}
+				} else if (stdoutBuf.length() == 4) {
+					stdoutExpecting = Utils::ntoh(*((const uint32_t *)stdoutBuf.data()));
+					stdoutBuf = "";
+					if (stdoutExpecting > ZT_SERVICE_MAX_MESSAGE_SIZE) {
+						LOG("message size overrun from service %s: %u bytes -- restarting service",_name.c_str(),stdoutExpecting);
+						stdoutExpecting = 0;
+						kill(_pid,SIGKILL);
+						break;
+					}
+				}
+			}
 		}
 	}
 }