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-lock.json

@@ -10,6 +10,7 @@
       "license": "AGPL-3.0",
       "dependencies": {
         "@influxdata/influxdb-client": "^1.20.0",
+        "@influxdata/influxdb-client-apis": "^1.20.0",
         "log4js": "^6.3.0",
         "luxon": "^2.1.1",
         "string-argv": "^0.3.1"
@@ -20,6 +21,14 @@
       "resolved": "https://registry.npmjs.org/@influxdata/influxdb-client/-/influxdb-client-1.20.0.tgz",
       "integrity": "sha512-jaKSI63hmQ5VSkJrFJkYIXaKlhoF+mGd4HmOf7v/X7pmEi69ReHp922Wyx6/OeCrpndRMbsadk+XmGNdd43cFw=="
     },
+    "node_modules/@influxdata/influxdb-client-apis": {
+      "version": "1.20.0",
+      "resolved": "https://registry.npmjs.org/@influxdata/influxdb-client-apis/-/influxdb-client-apis-1.20.0.tgz",
+      "integrity": "sha512-KMTmXH4rbpS+NWGpqDjxcKTyan2rbiT2IM5AdRElKhH2sHbH96xwLgziaxeC+OCJLeNAdehJgae3I8WiZjbwdg==",
+      "peerDependencies": {
+        "@influxdata/influxdb-client": "*"
+      }
+    },
     "node_modules/date-format": {
       "version": "3.0.0",
       "resolved": "https://registry.npmjs.org/date-format/-/date-format-3.0.0.tgz",
@@ -152,6 +161,12 @@
       "resolved": "https://registry.npmjs.org/@influxdata/influxdb-client/-/influxdb-client-1.20.0.tgz",
       "integrity": "sha512-jaKSI63hmQ5VSkJrFJkYIXaKlhoF+mGd4HmOf7v/X7pmEi69ReHp922Wyx6/OeCrpndRMbsadk+XmGNdd43cFw=="
     },
+    "@influxdata/influxdb-client-apis": {
+      "version": "1.20.0",
+      "resolved": "https://registry.npmjs.org/@influxdata/influxdb-client-apis/-/influxdb-client-apis-1.20.0.tgz",
+      "integrity": "sha512-KMTmXH4rbpS+NWGpqDjxcKTyan2rbiT2IM5AdRElKhH2sHbH96xwLgziaxeC+OCJLeNAdehJgae3I8WiZjbwdg==",
+      "requires": {}
+    },
     "date-format": {
       "version": "3.0.0",
       "resolved": "https://registry.npmjs.org/date-format/-/date-format-3.0.0.tgz",

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",
@@ -15,6 +15,7 @@
   "license": "AGPL-3.0",
   "dependencies": {
     "@influxdata/influxdb-client": "^1.20.0",
+    "@influxdata/influxdb-client-apis": "^1.20.0",
     "log4js": "^6.3.0",
     "luxon": "^2.1.1",
     "string-argv": "^0.3.1"

src/helper/influx-checks.js

@@ -0,0 +1,61 @@
+const logger = require.main.require("./helper/logger.js")("influx-checks");
+
+const Os = require("os");
+const { InfluxDB, Point } = require('@influxdata/influxdb-client')
+const Influx = require('@influxdata/influxdb-client-apis');
+
+
+function checkHealth(influxDb){
+    return new Promise((resolve, reject) => {
+        new Influx.HealthAPI(influxDb)    // Check influx health
+            .getHealth()
+            .catch((err) => {
+                logger.error("Could not communicate with Influx:");
+                logger.error(`Error [${err.code}]:`, err.message);
+                reject();
+            })
+            .then((res) => {
+                logger.debug("Server healthy.", "Version: ", res.version);
+                resolve(res);
+            });
+    });
+}
+
+function checkBucket(influxDb, options){
+    return new Promise((resolve, reject) => {
+        new Influx.BucketsAPI(influxDb).getBuckets(options)
+            .catch((err) => {       // Weirdly the influx-Api returns 404 for searches of non-existing buckets
+                logger.error("Could not get bucket:");
+                logger.error(`Error [${err.code}]:`, err.message);
+                reject();
+            }).then((res) => {      // But an empty list when the bucket exists, but token does not have permission to get details
+                logger.debug("Bucket found");
+                resolve(res);
+                // Now we know the bucket exists and we have some kind of permission.. but we still dont know if we are able to write to it..
+            });
+    });
+}
+
+function checkWriteApi(influxDb, options){
+    return new Promise((resolve, reject) => {
+        const writeApi = influxDb.getWriteApi(options.org, options.bucket);     // Get WriteAPI
+        writeApi.writePoint(new Point("worker_connectionTest").tag("hostname", Os.hostname()))    // Write point
+        writeApi.close()
+            .catch((err) => {
+                logger.error("Could not get writeApi:");
+                logger.error(`Error [${err.code}]:`, err.message);
+                reject();
+            }).then((res) => {
+                logger.debug("Writing ok");
+                resolve();
+            });
+    });
+}
+
+
+// Specify exports
+module.exports = {
+    checkHealth,
+    checkBucket,
+    checkWriteApi,
+};

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/helper/wifiStateAnalyzer.js

@@ -0,0 +1,43 @@
+const { HandshakeStage } = require.main.require('./dto/Packet.js');
+
+function keyInfoFromRaw(keyInfoRaw) {
+    return {
+        "KeyDescriptorVersion": keyInfoRaw>>0 & 0b111,
+        "KeyType":              keyInfoRaw>>3 & 0b1,
+        "KeyIndex":             keyInfoRaw>>4 & 0b11,
+        "Install":              keyInfoRaw>>6 & 0b1,
+        "KeyACK":               keyInfoRaw>>7 & 0b1,
+        "KeyMIC":               keyInfoRaw>>8 & 0b1,
+        "Secure":               keyInfoRaw>>9 & 0b1,
+        "Error":                keyInfoRaw>>10 & 0b1,
+        "Request":              keyInfoRaw>>11 & 0b1,
+        "EncryptedKeyData":     keyInfoRaw>>12 & 0b1,
+        "SMKMessage":           keyInfoRaw>>13 & 0b1,
+    };
+}
+
+const HANDSHAKE_STAGE_KEYINFO = {
+    "keys": ["Install", "KeyACK", "KeyMIC", "Secure"],
+    "0100": HandshakeStage[1],
+    "0010": HandshakeStage[2],
+    "1111": HandshakeStage[3],
+    "0011": HandshakeStage[4],
+};
+function handshakeStageFromKeyInfo(keyInfo){
+
+    // Extract compare-keys
+    let keyData = "";
+    for (const key of HANDSHAKE_STAGE_KEYINFO['keys']) {
+        keyData += keyInfo[key].toString();
+    }
+    
+    // Get and return stage
+    return  HANDSHAKE_STAGE_KEYINFO[keyData];
+}
+
+
+// Specify exports
+module.exports = {
+    keyInfoFromRaw,
+    handshakeStageFromKeyInfo,
+};

src/main.js

@@ -1,7 +1,21 @@
-const logger = require("./helper/logger.js")("main");
+"use strict";
+const logFactory = require("./helper/logger.js");
+const logger = logFactory("main");
 
 const { requireEnvVars } = require("./helper/env.js");
 const { exit } = require("process");
+const { exec } = require("./helper/exec.js");
+
+const { InfluxDB } = require('@influxdata/influxdb-client');
+const InfluxChecks = require('./helper/influx-checks.js');
+
+const { RegexBlockStream } = require("./streamHandler/RegexBlockStream.js");
+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;
@@ -20,3 +34,98 @@ if(errorMsg){
   exit(1);
 }
 
+(async function() {
+  logger.info("Setup Influx..");
+  const influxDb = new InfluxDB({url: env.INFLUX_URL, token: env.INFLUX_TOKEN});
+
+  await InfluxChecks.checkHealth(influxDb)
+    .then((res) => {return InfluxChecks.checkBucket(influxDb, {
+      org: env.INFLUX_ORG,
+      name: env.INFLUX_BUCKET
+    })})
+    .then((res) => {return InfluxChecks.checkWriteApi(influxDb, {
+      org: env.INFLUX_ORG,
+      bucket: env.INFLUX_BUCKET
+    })})
+    .catch((err) => {
+      if(err) {
+        logger.error("Error whilst checking influx:");
+        logger.error(err);
+      }
+      logger.fatal("Setup influx failed!");
+      exit(1);
+    });
+
+  logger.info("Influx ok");
+
+  logger.info("Starting tcpdump..");
+  const TCPDUMP_BASECMD = "tcpdump -vvv -e -n -X -s0 -i"
+  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(regexBlockStream)
+    .pipe(packetStreamFactory)
+    .pipe(packetInfluxPointFactory)
+    .pipe(influxPointWriter);
+
+  logger.debug("Attaching error-logger..");
+  const loggerTcpdump = logFactory("tcpdump");
+  proc.stderr.setEncoding("utf8").on("data", (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) => {
+    loggerTcpdump.debug(`tcpdump exited code: ${code}`);
+    if (code) {
+        loggerTcpdump.fatal(`tcpdump exited with non-zero code: ${code}`);
+        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/InfluxPointWriter.js

@@ -8,14 +8,16 @@ const {InfluxDB, Point, HttpError} = require('@influxdata/influxdb-client')
 class InfluxPointWriter extends Writable{
     /**
      * 
-     * @param {string} url Influx-Url
+     * @param {InfluxDB} influxDb InfluxDb
-     * @param {string} token Auth-token
      * @param {string} org Organization to use
      * @param {string} bucket Bucket to use
-     * @param {string} precision Precision to use
+     * @param {Partial<WriteOptions>} options Options for WriteApi
      */
-    constructor(url, token, org, bucket, precision = 'us'){
+    constructor(influxDb, org, bucket, options){
-        this._api = new InfluxDB({url, token}).getWriteApi(org, bucket, precision);
+        super({
+            objectMode: true
+        });
+        this._api = influxDb.getWriteApi(org, bucket, 'us', options);
     }
 
     _write(point, encoding, next){
@@ -25,10 +27,8 @@ class InfluxPointWriter extends Writable{
 
     _flush(next){
         this._api.flush(true)
-            .then(
+            .catch((err) => { next(new Error(`WriteApi rejected promise for flush: ${err}`)); })
-                next,
+            .then(next);
-                (err) => { next(new Error(`WriteApi rejected promise for flush: ${err}`)); }
-            );
     }
 }
 

src/streamHandler/PacketInfluxPointFactory.js

@@ -9,6 +9,7 @@ const TAG_LIST = [
     "bssid",
     "frequency",
     "flags",
+    "packetType",
 ];
 
 /** Measurement-name and corresponding field-key */
@@ -37,23 +38,35 @@ class PacketInfluxPointFactory extends Transform{
     _transform(packet, encoding, next){
         // Create measurements
         MEASUREMENT_MAP.forEach((objKey, measurement) => {
-            if(!Object.keys(packet).includes(objKey)) return;
+            if(packet[objKey] == null) return;
 
             let point = new Point(measurement);     // Create point
             
             // Set tags
-            TAG_LIST.filter(tag => Object.keys(packet).includes(tag))
+            TAG_LIST.filter(tag => Object.keys(packet).includes(tag))   // Filter tags available on object
-                    .forEach(tag => point.tag(tag, packet[tag]));
+                    .filter(tag => packet[tag] != null)                         // Filter tags not falsy on object
+                    .forEach(tag => {
+                        tagObjectRecursively(point, tag, packet[tag]);
+                    });
 
             point.setField('value', packet[objKey]);        // Set field
 
-            this.push(null, point);     // Push point into stream
+            this.push(point);     // Push point into stream
         });
         
         next();     // Get next packet
     }
 }
 
+function tagObjectRecursively(point, tag, field, suffix = ""){
+    if(typeof(field) == "object"){
+        // TODO: Convert boolean-arrays like "packet.flags" to key: value
+        Object.entries(field).map(([key, value]) => {
+            tagObjectRecursively(point, tag, value, `_${key}${suffix}`);
+        });
+    }
+    else point.tag(tag+suffix, field);
+}
 
 /** Mapping for type -> field-method */
 const POINT_FIELD_TYPE = new Map([

src/streamHandler/PacketStreamFactory.js

@@ -3,6 +3,7 @@ const { Transform } = require('stream');
 const { DateTime } = require("luxon");
 const { PacketType, FlagType, Packet, PacketWithSSID, BeaconPacket, ProbeRequestPacket, ProbeResponsePacket, AuthenticationPacket, AuthenticationType, AssociationResponsePacket, DisassociationPacket, HandshakePacket, HandshakeStage } = require.main.require('./dto/Packet.js');
 const hexConv = require.main.require("./helper/hexConverter.js");
+const wifiStateAnalyser = require.main.require("./helper/wifiStateAnalyzer.js");
 
 const PACKET_TYPE_MAP = {
     "Beacon":           PacketType.Beacon,
@@ -65,17 +66,22 @@ class PacketStreamFactory extends Transform{
             ?.forEach(match => packet.flags[FLAG_TYPE_MAP[match]] = true)       // Set them to true in flags
         );
 
-        packet.dataRate = Number(data.match(/(?<=^|\s)[0-9]+(\.[0-9]+)?(?=\sMb\/?s($|\s))/i)?.[0]) || null;
+        packet.dataRate = Number(data.match(/(?<=^|\s)\d+(\.\d+)?(?=\sMb\/?s($|\s))/i)?.[0]) || null;
-        packet.frequency = Number(data.match(/(?<=^|\s)[0-9]{4}(?=\sMHz($|\s))/i)?.[0]) || null;
+        packet.frequency = Number(data.match(/(?<=^|\s)\d{4}(?=\sMHz($|\s))/i)?.[0]) || null;
 
-        packet.durationMicros = Number(data.match(/(?<=^|\s)[0-9]{1,4}(?=us($|\s))/i)?.[0]) || null;
+        packet.durationMicros = Number(data.match(/(?<=^|\s)\d{1,4}(?=us($|\s))/i)?.[0]) || null;
 
-        packet.signal = Number(data.match(/(?<=^|\s)-[0-9]{2,3}(?=dBm\sSignal($|\s))/i)?.[0]) || null;
+        packet.signal = Number(data.match(/(?<=^|\s)-\d{2,3}(?=dBm\sSignal($|\s))/i)?.[0]) || null;
 
         let packetTypeStr = data.match(new RegExp('(?<=^|\\s)('+ PACKET_TYPES_REGEX +')(?=$|\\s)', 'i'))?.[0];
-        packet.packetType = packetTypeStr? PACKET_TYPE_MAP[packetTypeStr]: 
+        if(packetTypeStr)
-                            data.match(/(SA|TA|DA|RA|BSSID):.{17}\s*$/i)? PacketType.NoData:
+            packet.packetType = PACKET_TYPE_MAP[packetTypeStr];
-                            PacketType.Unknown;
+        else if(data.match(/(SA|TA|DA|RA|BSSID):.{17}\s*$/i)){
+            packet.packetType = PacketType.NoData
+        }
+        else {
+            packet.packetType = PacketType.Unknown;
+        }
 
         packet.srcMac = data.match(/(?<=(^|\s)(SA|TA):).{17}(?=$|\s)/i)?.[0] ?? null;
 
@@ -129,25 +135,9 @@ class PacketStreamFactory extends Transform{
 
                 // Read key-information
                 const keyInfoRaw = (packet.payloadData[0x5]<<0x8) + packet.payloadData[0x6];
-                const keyInfo = {
+                const keyInfo = wifiStateAnalyser.keyInfoFromRaw(keyInfoRaw);   // Convert
-                    "KeyDescriptorVersion": keyInfoRaw>>0 & 0b111,
-                    "KeyType":  keyInfoRaw>>3 & 0b1,
-                    "KeyIndex": keyInfoRaw>>4 & 0b11,
-                    "Install":  keyInfoRaw>>6 & 0b1,
-                    "KeyACK":   keyInfoRaw>>7 & 0b1,
-                    "KeyMIC":   keyInfoRaw>>8 & 0b1,
-                    "Secure":   keyInfoRaw>>9 & 0b1,
-                    "Error":    keyInfoRaw>>10 & 0b1,
-                    "Request":  keyInfoRaw>>11 & 0b1,
-                    "EncryptedKeyData": keyInfoRaw>>12 & 0b1,
-                    "SMKMessage":       keyInfoRaw>>13 & 0b1,
-                };
 
-                newPacket.handshakeStage =  (!keyInfo.Install &&  keyInfo.KeyACK && !keyInfo.KeyMIC && !keyInfo.Secure)? HandshakeStage[1] :
+                newPacket.handshakeStage =  wifiStateAnalyser.handshakeStageFromKeyInfo(keyInfo);   // Get stage
-                                            (!keyInfo.Install && !keyInfo.KeyACK &&  keyInfo.KeyMIC && !keyInfo.Secure)? HandshakeStage[2] :
-                                            ( keyInfo.Install &&  keyInfo.KeyACK &&  keyInfo.KeyMIC &&  keyInfo.Secure)? HandshakeStage[3] :
-                                            (!keyInfo.Install && !keyInfo.KeyACK &&  keyInfo.KeyMIC &&  keyInfo.Secure)? HandshakeStage[4] :
-                                            null;
                 break;
         }
         if(newPacket) packet = Object.assign(newPacket, packet);

src/streamHandler/RegexBlockStream.js

@@ -13,7 +13,7 @@ class RegexBlockStream extends Transform{
      * @param {RegExp} matcher Block-match
      * @param {boolean} withholdLastBlock When true, the last matches block will not be submitted to prevent submitting incomplete blocks.
      * @param {boolean} matchAllOnFlush (Only in combination with withholdLastBlock) When enabled, the buffer will be matched on last time on _flush (stream deconstruction) and write any, also incomplete, blocks
-     * @remarks WARNING: It should match a clean-block (including e.g. newline)! Otherwise buffer will get dirty and use more and more ressources.
+     * @remarks WARNING: It should match a clean-block (including e.g. newline)! Otherwise buffer will get dirty and use more and more resources.
      */
     constructor(matcher, withholdLastBlock = true, matchAllOnFlush = false){
         super({
@@ -27,7 +27,7 @@ class RegexBlockStream extends Transform{
     }
 
     _transform(chunk, encoding, next){
-        chunk = this.readableBuffer.length? this.readableBuffer.join() + chunk: chunk;     // Add previous buffer to current chunk
+        chunk = this.readableBuffer.length? this.readableBuffer.join('') + chunk: chunk;     // Add previous buffer to current chunk
         this.readableBuffer.length && this.readableBuffer.clear();      // Clear buffer once we read it
 
         let matches = chunk.match(this.matcher);    // Match
@@ -52,7 +52,7 @@ class RegexBlockStream extends Transform{
 
     _flush(next){
         if(matchAllOnFlush){    // When requested, we'll match one last time over the remaining buffer
-            let chunk = this.readableBuffer.toString();
+            let chunk = this.readableBuffer.join('');
             let matches = chunk.match(this.matcher);    // Match remaining buffer
             _writeMatches(matches);    // Write matches including last element
         }