Merge branch 'release-1.0'

tcpdump writes to stderr at the beginning

Dockerfile

@@ -7,7 +7,11 @@ WORKDIR /usr/src/app
 COPY package*.json ./
 RUN npm install
 
+RUN apt-get update
+RUN apt-get -y install \
+    tcpdump
+
 # Bundle app source
 COPY ./src/ .
 
-CMD ["npm", "run"]
+CMD ["npm", "run", "start"]

package.json

@@ -2,10 +2,10 @@
   "name": "rfmon-to-influx",
   "version": "1.0.0",
   "description": "Writing (mostly meta-) data received in Wireless-Monitor-Mode into an InfluxDB",
-  "main": "src/main.js",
+  "main": "main.js",
   "scripts": {
     "test": "echo \"Error: no test specified\" && exit 1",
-    "start": "node src/main.js"
+    "start": "node main.js"
   },
   "repository": {
     "type": "git",

src/helper/userHelper.js

@@ -0,0 +1,46 @@
+// This file specifies functions to help a user with e.g. configuration-errors
+
+function detectStreamData(stream, timeout = 5000){
+    return new Promise((resolve, reject) => {
+        let timeoutHandler;
+        if(timeout){
+            timeoutHandler = setTimeout(() => {
+                reject('timeout');
+                remListeners();
+            },
+            timeout);
+        }
+
+        function remListeners(){ 
+            stream.removeListener('error', errorHandler);
+            stream.removeListener('data', dataHandler);
+            if(timeoutHandler) clearTimeout(timeoutHandler);
+        }
+
+        function errorHandler(err) {
+            remListeners();
+        }
+        function dataHandler(data) {
+            resolve(data);
+            remListeners();
+        }
+
+        stream.on('error', errorHandler);
+        stream.on('data', dataHandler);
+    });    
+}
+
+function detectStreamsData(streams, timeout = 5000){
+    let promises = [];
+    streams.forEach((stream) => {
+        promises.push(detectStreamData(stream, timeout));
+    })
+    return promises;
+}
+
+
+// Specify exports
+module.exports = {
+    detectStreamData,
+    detectStreamsData,
+};

src/main.js

@@ -1,5 +1,6 @@
 "use strict";
-const logger = require("./helper/logger.js")("main");
+const logFactory = require("./helper/logger.js");
+const logger = logFactory("main");
 
 const { requireEnvVars } = require("./helper/env.js");
 const { exit } = require("process");
@@ -13,6 +14,9 @@ const { PacketStreamFactory } = require("./streamHandler/PacketStreamFactory.js"
 const { PacketInfluxPointFactory } = require("./streamHandler/PacketInfluxPointFactory.js");
 const { InfluxPointWriter } = require("./streamHandler/InfluxPointWriter.js");
 
+const userHelper = require("./helper/userHelper.js");
+
+
 /// Setup ENVs
 const env = process.env;
 // Defaults
@@ -56,32 +60,72 @@ if(errorMsg){
 
   logger.info("Starting tcpdump..");
   const TCPDUMP_BASECMD = "tcpdump -vvv -e -n -X -s0 -i"
-  let cmd = `sudo ${TCPDUMP_BASECMD} ${env.WIFI_INTERFACE}`;
+  let cmd = `${TCPDUMP_BASECMD} ${env.WIFI_INTERFACE}`;
 
   let proc = exec(cmd);
   logger.debug("Creating & Attaching streams..");
+  let regexBlockStream = new RegexBlockStream(/^\d{2}:\d{2}:\d{2}.\d{6}.*(\n( {4,8}|\t\t?).*)+\n/gm);
+  let packetStreamFactory = new PacketStreamFactory();
+  let packetInfluxPointFactory = new PacketInfluxPointFactory();
+  let influxPointWriter = new InfluxPointWriter(influxDb, env.INFLUX_ORG, env.INFLUX_BUCKET);
   proc.stdout
     .setEncoding("utf8")
-    .pipe(new RegexBlockStream(/^\d{2}:\d{2}:\d{2}.\d{6}.*(\n( {4,8}|\t\t?).*)+\n/gm))
+    .pipe(regexBlockStream)
-    .pipe(new PacketStreamFactory())
+    .pipe(packetStreamFactory)
-    .pipe(new PacketInfluxPointFactory())
+    .pipe(packetInfluxPointFactory)
-    .pipe(new InfluxPointWriter(influxDb, env.INFLUX_ORG, env.INFLUX_BUCKET));
+    .pipe(influxPointWriter);
 
   logger.debug("Attaching error-logger..");
+  const loggerTcpdump = logFactory("tcpdump");
   proc.stderr.setEncoding("utf8").on("data", (data) => {
-    logger.error(data);
+    if(!data.match(/^(tcpdump: )?listening on /i) || !data.match(/^\d+ packets captured/i)) {  // Catch start-error
+        loggerTcpdump.debug(data);
+    }
+    else loggerTcpdump.error(data);
   });
 
+  // FIXME: This is a hacky workaround to not let errors from subprocess bubble up and terminate our process
+  regexBlockStream.on('error', (err) => {});
+
+  proc.on("error", (err) => {
+    loggerTcpdump.error(err);
+  });
+
+  const loggerPacketStream = logFactory("PacketStreamFactory");
+  userHelper.detectStreamData(proc.stdout, 10000)       // Expect tcpdump-logs to have data after max. 10s
+    .then(() => {
+        loggerTcpdump.debug("Got first data");
+        userHelper.detectStreamData(packetStreamFactory, 10000)      // Expect then to have packets after further 10s
+            .then(() => {
+                loggerPacketStream.debug("Got first packet");
+            })
+            .catch((err) => {
+                if(err == 'timeout') loggerPacketStream.warn("No packets");
+            });
+    })
+    .catch((err) => {
+        if(err == 'timeout') loggerTcpdump.warn("No data after 10s! Wrong configuration?");
+    });
+
   logger.debug("Attaching exit-handler..");
   proc.on("exit", (code) => {
-    logger.info(`tcpdump exited code: ${code}`);
+    loggerTcpdump.debug(`tcpdump exited code: ${code}`);
     if (code) {
-      logger.fatal(`tcpdump exited with non-zero code: ${code}`);
+        loggerTcpdump.fatal(`tcpdump exited with non-zero code: ${code}`);
-      exit(1);
+        exit(1);
     }
     logger.info("Shutdown");
     exit(0);
   });
 
+  // Handle stop-signals for graceful shutdown
+  function shutdownReq() {
+    logger.info("Shutdown request received..");
+    logger.debug("Stopping subprocess tcpdump, then exiting myself..");
+    proc.kill();    // Kill process (send SIGTERM), then upper event-handler will stop self
+  }
+  process.on('SIGTERM', shutdownReq);
+  process.on('SIGINT', shutdownReq);
+
   logger.info("Startup complete");
 })();

src/streamHandler/PacketInfluxPointFactory.js

@@ -9,6 +9,7 @@ const TAG_LIST = [
     "bssid",
     "frequency",
     "flags",
+    "packetType",
 ];
 
 /** Measurement-name and corresponding field-key */