16 Commits

5 changed files with 98 additions and 13 deletions

View File

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

View File

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

46
src/helper/userHelper.js Normal file
View File

@@ -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,
};

View File

@@ -1,5 +1,6 @@
"use strict"; "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 { requireEnvVars } = require("./helper/env.js");
const { exit } = require("process"); const { exit } = require("process");
@@ -13,6 +14,9 @@ const { PacketStreamFactory } = require("./streamHandler/PacketStreamFactory.js"
const { PacketInfluxPointFactory } = require("./streamHandler/PacketInfluxPointFactory.js"); const { PacketInfluxPointFactory } = require("./streamHandler/PacketInfluxPointFactory.js");
const { InfluxPointWriter } = require("./streamHandler/InfluxPointWriter.js"); const { InfluxPointWriter } = require("./streamHandler/InfluxPointWriter.js");
const userHelper = require("./helper/userHelper.js");
/// Setup ENVs /// Setup ENVs
const env = process.env; const env = process.env;
// Defaults // Defaults
@@ -56,27 +60,57 @@ if(errorMsg){
logger.info("Starting tcpdump.."); logger.info("Starting tcpdump..");
const TCPDUMP_BASECMD = "tcpdump -vvv -e -n -X -s0 -i" 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); let proc = exec(cmd);
logger.debug("Creating & Attaching streams.."); 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 proc.stdout
.setEncoding("utf8") .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.."); logger.debug("Attaching error-logger..");
const loggerTcpdump = logFactory("tcpdump");
proc.stderr.setEncoding("utf8").on("data", (data) => { proc.stderr.setEncoding("utf8").on("data", (data) => {
logger.error(data); if(!data.match(/^(tcpdump: )?listening on /i)) // Catch start-error
loggerTcpdump.error(data);
});
regexBlockStream.on('error', (err) => {
if(err) loggerTcpdump.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.."); logger.debug("Attaching exit-handler..");
proc.on("exit", (code) => { proc.on("exit", (code) => {
logger.info(`tcpdump exited code: ${code}`); loggerTcpdump.debug(`tcpdump exited code: ${code}`);
if (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"); logger.info("Shutdown");

View File

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