Netconf service itself works, time to integrate.

This commit is contained in:
Adam Ierymenko 2013-08-02 14:25:23 -04:00
parent 741642ba53
commit 2a6b74746e
6 changed files with 107 additions and 51 deletions

View file

@ -1,6 +1,7 @@
all: all:
gcc -O6 -c ../ext/lz4/lz4hc.c ../ext/lz4/lz4.c 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 g++ -DZT_OSNAME="linux" -DZT_ARCH="x86_64" -I/usr/include/mysql -I../ext/bin/libcrypto/include -O -pthread -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
g++ -DZT_OSNAME="linux" -DZT_ARCH="x86_64" -I/usr/include/mysql -I../ext/bin/libcrypto/include -O -pthread -o netconf-test netconf-test.cpp ../node/Utils.cpp ../node/Identity.cpp ../node/EllipticCurveKeyPair.cpp ../node/Salsa20.cpp ../node/HMAC.cpp ../node/Logger.cpp ../node/Service.cpp ../node/Thread.cpp lz4.o lz4hc.o ../ext/bin/libcrypto/linux-x86_64/libcrypto.a
clean: clean:
rm -f *.o netconf.service rm -f *.o netconf.service netconf-test

View file

@ -42,11 +42,13 @@
#include "../node/Identity.hpp" #include "../node/Identity.hpp"
#include "../node/RuntimeEnvironment.hpp" #include "../node/RuntimeEnvironment.hpp"
#include "../node/Logger.hpp" #include "../node/Logger.hpp"
#include "../node/Thread.hpp"
using namespace ZeroTier; using namespace ZeroTier;
static void svcHandler(void *arg,Service &svc,const Dictionary &msg) static void svcHandler(void *arg,Service &svc,const Dictionary &msg)
{ {
std::cout << msg.toString();
} }
int main(int argc,char **argv) int main(int argc,char **argv)
@ -59,8 +61,23 @@ int main(int argc,char **argv)
std::vector<Identity> population; std::vector<Identity> population;
for(;;) { for(;;) {
Identity id;
if ((population.empty())||(rand() < (RAND_MAX / 4))) { if ((population.empty())||(rand() < (RAND_MAX / 4))) {
id.generate();
population.push_back(id);
std::cout << "Testing with new identity: " << id.address().toString() << std::endl;
} else { } else {
id = population[rand() % population.size()];
Thread::sleep(1000);
std::cout << "Testing with existing identity: " << id.address().toString() << std::endl;
} }
Dictionary request;
request["type"] = "netconf-request";
request["peerId"] = id.toString(false);
request["nwid"] = "6c92786fee000001";
request["requestId"] = "12345";
svc.send(request);
} }
} }

View file

@ -55,6 +55,7 @@
#include <string.h> #include <string.h>
#include <stdint.h> #include <stdint.h>
#include <unistd.h> #include <unistd.h>
#include <errno.h>
#include <sys/stat.h> #include <sys/stat.h>
#include <sys/types.h> #include <sys/types.h>
#include <arpa/inet.h> #include <arpa/inet.h>
@ -71,24 +72,34 @@
#include "../node/Dictionary.hpp" #include "../node/Dictionary.hpp"
#include "../node/Identity.hpp" #include "../node/Identity.hpp"
#include "../node/Utils.hpp" #include "../node/Utils.hpp"
#include "../node/Mutex.hpp"
using namespace ZeroTier; using namespace ZeroTier;
using namespace mysqlpp; using namespace mysqlpp;
static Mutex stdoutWriteLock;
static Connection *dbCon = (Connection *)0; static Connection *dbCon = (Connection *)0;
static char mysqlHost[64],mysqlPort[64],mysqlDatabase[64],mysqlUser[64],mysqlPassword[64]; static char mysqlHost[64],mysqlPort[64],mysqlDatabase[64],mysqlUser[64],mysqlPassword[64];
static void connectOrReconnect() static void connectOrReconnect()
{ {
for(;;) { for(;;) {
if (dbCon)
delete dbCon; delete dbCon;
try {
dbCon = new Connection(mysqlDatabase,mysqlHost,mysqlUser,mysqlPassword,(unsigned int)strtol(mysqlPort,(char **)0,10)); dbCon = new Connection(mysqlDatabase,mysqlHost,mysqlUser,mysqlPassword,(unsigned int)strtol(mysqlPort,(char **)0,10));
if (dbCon->connected()) if (dbCon->connected()) {
fprintf(stderr,"(re?)-connected to mysql server successfully\n");
break; break;
else { } else {
fprintf(stderr,"Unable to connect to database server.\n"); fprintf(stderr,"unable to connect to database server (connection closed), trying again in 1s...\n");
usleep(1000); usleep(1000000);
}
} catch (std::exception &exc) {
fprintf(stderr,"unable to connect to database server (%s), trying again in 1s...\n",exc.what());
usleep(1000000);
} catch ( ... ) {
fprintf(stderr,"unable to connect to database server (unknown exception), trying again in 1s...\n");
usleep(1000000);
} }
} }
} }
@ -98,7 +109,7 @@ int main(int argc,char **argv)
{ {
char *ee = getenv("ZT_NETCONF_MYSQL_HOST"); char *ee = getenv("ZT_NETCONF_MYSQL_HOST");
if (!ee) { if (!ee) {
fprintf(stderr,"Missing environment variable: ZT_NETCONF_MYSQL_HOST\n"); fprintf(stderr,"missing environment variable: ZT_NETCONF_MYSQL_HOST\n");
return -1; return -1;
} }
strcpy(mysqlHost,ee); strcpy(mysqlHost,ee);
@ -108,57 +119,66 @@ int main(int argc,char **argv)
else strcpy(mysqlPort,ee); else strcpy(mysqlPort,ee);
ee = getenv("ZT_NETCONF_MYSQL_DATABASE"); ee = getenv("ZT_NETCONF_MYSQL_DATABASE");
if (!ee) { if (!ee) {
fprintf(stderr,"Missing environment variable: ZT_NETCONF_MYSQL_DATABASE\n"); fprintf(stderr,"missing environment variable: ZT_NETCONF_MYSQL_DATABASE\n");
return -1; return -1;
} }
strcpy(mysqlDatabase,ee); strcpy(mysqlDatabase,ee);
ee = getenv("ZT_NETCONF_MYSQL_USER"); ee = getenv("ZT_NETCONF_MYSQL_USER");
if (!ee) { if (!ee) {
fprintf(stderr,"Missing environment variable: ZT_NETCONF_MYSQL_USER\n"); fprintf(stderr,"missing environment variable: ZT_NETCONF_MYSQL_USER\n");
return -1; return -1;
} }
strcpy(mysqlUser,ee); strcpy(mysqlUser,ee);
ee = getenv("ZT_NETCONF_MYSQL_PASSWORD"); ee = getenv("ZT_NETCONF_MYSQL_PASSWORD");
if (!ee) { if (!ee) {
fprintf(stderr,"Missing environment variable: ZT_NETCONF_MYSQL_PASSWORD\n"); fprintf(stderr,"missing environment variable: ZT_NETCONF_MYSQL_PASSWORD\n");
return -1; return -1;
} }
strcpy(mysqlPassword,ee); strcpy(mysqlPassword,ee);
} }
char buf[4096]; char buf[131072];
std::string dictBuf; std::string dictBuf;
connectOrReconnect(); connectOrReconnect();
for(;;) { for(;;) {
if (read(STDIN_FILENO,buf,4) != 4) { for(int l=0;l<4;) {
fprintf(stderr,"Error reading frame size from stdin\n"); int n = (int)read(STDIN_FILENO,buf + l,4 - l);
if (n < 0) {
fprintf(stderr,"error reading frame size from stdin: %s\n",strerror(errno));
return -1; return -1;
} }
l += n;
}
unsigned int fsize = (unsigned int)ntohl(*((const uint32_t *)buf)); unsigned int fsize = (unsigned int)ntohl(*((const uint32_t *)buf));
while (dictBuf.length() < fsize) { while (dictBuf.length() < fsize) {
int n = (int)read(STDIN_FILENO,buf,std::min((int)sizeof(buf),(int)(fsize - dictBuf.length()))); int n = (int)read(STDIN_FILENO,buf,std::min((int)sizeof(buf),(int)(fsize - dictBuf.length())));
if (n < 0) {
fprintf(stderr,"error reading frame from stdin: %s\n",strerror(errno));
return -1;
}
for(int i=0;i<n;++i) for(int i=0;i<n;++i)
dictBuf.push_back(buf[i]); dictBuf.push_back(buf[i]);
} }
Dictionary msg(dictBuf); Dictionary request(dictBuf);
dictBuf = ""; dictBuf = "";
if (!dbCon->connected()) if (!dbCon->connected())
connectOrReconnect(); connectOrReconnect();
try { try {
const std::string &command = msg.get("command"); const std::string &reqType = request.get("type");
if (command == "config") { // NETWORK_CONFIG_REQUEST packet if (reqType == "netconf-request") { // NETWORK_CONFIG_REQUEST packet
Identity peerIdentity(msg.get("peerIdentity")); Identity peerIdentity(request.get("peerId"));
uint64_t nwid = strtoull(msg.get("nwid").c_str(),(char **)0,16); uint64_t nwid = strtoull(request.get("nwid").c_str(),(char **)0,16);
Dictionary meta; Dictionary meta;
if (msg.contains("meta")) if (request.contains("meta"))
meta.fromString(msg.get("meta")); meta.fromString(request.get("meta"));
// Do quick signature check / sanity check // Do quick signature check / sanity check
if (!peerIdentity.locallyValidate(false)) { if (!peerIdentity.locallyValidate(false)) {
fprintf(stderr,"identity failed signature check: %s",peerIdentity.toString(false).c_str()); fprintf(stderr,"identity failed signature check: %s\n",peerIdentity.toString(false).c_str());
continue; continue;
} }
@ -176,16 +196,22 @@ int main(int argc,char **argv)
} }
} else { } else {
q = dbCon->query(); q = dbCon->query();
uint64_t now = Utils::now(); q << "INSERT INTO Node (id,creationTime,lastSeen,identity) VALUES (" << peerIdentity.address().toInt() << "," << Utils::now() << ",0," << quote << peerIdentity.toString(false) << ")";
q << "INSERT INTO Node (id,creationTime,lastSeen,identity) VALUES (" << peerIdentity.address().toInt() << "," << now << "," << now << "," << peerIdentity.toString(false) << ")";
if (!q.exec()) { if (!q.exec()) {
fprintf(stderr,"Error inserting Node row for peer %s, aborting netconf request",peerIdentity.address().toString().c_str()); fprintf(stderr,"error inserting Node row for peer %s, aborting netconf request\n",peerIdentity.address().toString().c_str());
continue; continue;
} }
// TODO: launch background validation // TODO: launch background validation
} }
} }
// Update lastSeen
{
Query q = dbCon->query();
q << "UPDATE Node SET lastSeen = " << Utils::now() << " WHERE id = " << peerIdentity.address().toInt();
q.exec();
}
bool isOpen = false; bool isOpen = false;
{ {
Query q = dbCon->query(); Query q = dbCon->query();
@ -278,20 +304,26 @@ int main(int argc,char **argv)
if (ipv6Static.length()) if (ipv6Static.length())
netconf["ipv6Static"] = ipv6Static; netconf["ipv6Static"] = ipv6Static;
Dictionary resp; {
resp["peer"] = peerIdentity.address().toString(); Dictionary response;
resp["nwid"] = msg.get("nwid"); response["peer"] = peerIdentity.address().toString();
resp["requestId"] = msg.get("requestId"); response["nwid"] = request.get("nwid");
resp["netconf"] = netconf.toString(); response["type"] = "netconf-response";
std::string respm = resp.toString(); response["requestId"] = request.get("requestId");
response["netconf"] = netconf.toString();
std::string respm = response.toString();
uint32_t respml = (uint32_t)htonl((uint32_t)respm.length()); uint32_t respml = (uint32_t)htonl((uint32_t)respm.length());
stdoutWriteLock.lock();
write(STDOUT_FILENO,&respml,4); write(STDOUT_FILENO,&respml,4);
write(STDOUT_FILENO,respm.data(),respm.length()); write(STDOUT_FILENO,respm.data(),respm.length());
stdoutWriteLock.unlock();
}
} }
} catch (std::exception &exc) { } catch (std::exception &exc) {
fprintf(stderr,"unexpected exception handling message: %s",exc.what()); fprintf(stderr,"unexpected exception handling message: %s\n",exc.what());
} catch ( ... ) { } catch ( ... ) {
fprintf(stderr,"unexpected exception handling message: unknown exception"); fprintf(stderr,"unexpected exception handling message: unknown exception\n");
} }
} }
} }

View file

@ -104,7 +104,7 @@ public:
_keyPair((EllipticCurveKeyPair *)0) _keyPair((EllipticCurveKeyPair *)0)
{ {
if (!fromString(str)) if (!fromString(str))
throw std::invalid_argument("invalid string-serialized identity"); throw std::invalid_argument(std::string("invalid string-serialized identity: ") + str);
} }
Identity(const std::string &str) Identity(const std::string &str)
@ -112,7 +112,7 @@ public:
_keyPair((EllipticCurveKeyPair *)0) _keyPair((EllipticCurveKeyPair *)0)
{ {
if (!fromString(str)) if (!fromString(str))
throw std::invalid_argument("invalid string-serialized identity"); throw std::invalid_argument(std::string("invalid string-serialized identity: ") + str);
} }
template<unsigned int C> template<unsigned int C>

View file

@ -36,6 +36,7 @@
#include <signal.h> #include <signal.h>
#include <time.h> #include <time.h>
#include <fcntl.h> #include <fcntl.h>
#include <errno.h>
#include <sys/time.h> #include <sys/time.h>
#include <sys/types.h> #include <sys/types.h>
#include <sys/stat.h> #include <sys/stat.h>
@ -109,7 +110,7 @@ bool Service::send(const Dictionary &msg)
void Service::main() void Service::main()
throw() throw()
{ {
char buf[4096]; char buf[131072];
fd_set readfds,writefds,exceptfds; fd_set readfds,writefds,exceptfds;
struct timeval tv; struct timeval tv;
@ -126,27 +127,30 @@ void Service::main()
pipe(out); pipe(out);
pipe(err); pipe(err);
long pid = fork(); long pid = vfork();
if (pid < 0) { if (pid < 0) {
LOG("service %s terminating: could not fork!",_name.c_str()); LOG("service %s terminating: could not fork!",_name.c_str());
return; return;
} else if (pid) { } else if (pid) {
close(in[1]); // Parent
close(out[0]); close(in[0]);
close(err[0]); close(out[1]);
close(err[1]);
Thread::sleep(500); // give child time to start Thread::sleep(500); // give child time to start
_childStdin = in[1]; _childStdin = in[1];
_childStdout = out[0]; _childStdout = out[0];
_childStderr = err[0]; _childStderr = err[0];
fcntl(_childStdout,F_SETFL,O_NONBLOCK); fcntl(_childStdout,F_SETFL,O_NONBLOCK);
fcntl(_childStderr,F_SETFL,O_NONBLOCK); fcntl(_childStderr,F_SETFL,O_NONBLOCK);
_pid = pid;
} else { } else {
dup2(in[0],STDIN_FILENO); // Child
dup2(out[1],STDOUT_FILENO);
dup2(err[1],STDERR_FILENO);
close(in[1]); close(in[1]);
close(out[0]); close(out[0]);
close(err[0]); close(err[0]);
dup2(in[0],STDIN_FILENO);
dup2(out[1],STDOUT_FILENO);
dup2(err[1],STDERR_FILENO);
execl(_path.c_str(),_path.c_str(),_r->homePath.c_str(),(const char *)0); execl(_path.c_str(),_path.c_str(),_r->homePath.c_str(),(const char *)0);
exit(-1); exit(-1);
} }
@ -169,6 +173,8 @@ void Service::main()
} }
} }
// If we've made it here, _pid is running last we checked.
FD_ZERO(&readfds); FD_ZERO(&readfds);
FD_ZERO(&writefds); FD_ZERO(&writefds);
FD_ZERO(&exceptfds); FD_ZERO(&exceptfds);

View file

@ -75,7 +75,7 @@ void Thread::join()
void Thread::sleep(unsigned long ms) void Thread::sleep(unsigned long ms)
{ {
usleep(ms); usleep(ms * 1000);
} }
void Thread::__intl_run() void Thread::__intl_run()