(features) Add basic express ingestor

Author: AHarmlessPyro

ingestors/express/package.json

@@ -0,0 +1,20 @@
+{
+  "name": "@metlo/express-ingest",
+  "description": "An ingestor for metlo api-security to support the express server",
+  "version": "1.0.0",
+  "main": "index.js",
+  "license": "MIT",
+  "scripts": {},
+  "dependencies": {
+    "express": "^4.0.0",
+    "shimmer": "^1.2.1"
+  },
+  "repository": {
+    "type": "git",
+    "url": "https://github.com/metlo-labs/metlo.git"
+  },
+  "engines": {
+    "node": ">=11.7.0"
+  },
+  "homepage": "www.metlo.com"
+}

ingestors/express/src/index.js

@@ -0,0 +1,78 @@
+var shimmer = require('shimmer');
+var expr_resp = require('express/lib/response');
+var http = require('https')
+var os = require('node:os');
+
+var WorkerPool = require('./pool');
+
+
+const pool = new WorkerPool(os.cpus().length, './workerTarget.js');
+
+function exit() {
+    pool.close()
+}
+
+process.on('exit', exit);
+process.on('SIGTERM', exit);
+
+
+var metloDetails = {
+    key: null,
+    host: null
+}
+
+
+function compileInformation(_req, _res, responseBody) {
+    const data = JSON.stringify({
+        request: {
+            url: {
+                host: _req.hostname,
+                path: _req.route.path,
+                parameters: _req.query,
+            },
+            headers: _req.headers,
+            body: _req.body || "No Body",
+            method: _req.method,
+        },
+        response: {
+            url: `${_req.socket.remoteAddress}:${_req.socket.remotePort}`,
+            status: _res.statusCode,
+            headers: _res.getHeaders(),
+            body: responseBody,
+        },
+        meta: {
+            environment: "production",
+            incoming: true,
+            source: _req.socket.remoteAddress,
+            sourcePort: _req.socket.remotePort,
+            // TODO : Add destination
+            destination: "server.hostname",
+            destinationPort: "server.port",
+        }
+    })
+
+    pool.runTask({ host: metloDetails.host, key: metloDetails.key, data }, (err, result) => {
+        console.log(err, result);
+    })
+}
+
+shimmer.wrap(expr_resp, 'send', function (original) {
+    return function () {
+        var returned = original.apply(this, arguments)
+        compileInformation(this.req, returned, arguments[arguments.length - 1])
+        return returned;
+    };
+});
+
+shimmer.wrap(expr_resp, 'sendFile', function (original) {
+    return function () {
+        var returned = original.apply(this, arguments)
+        compileInformation(this.req, returned, JSON.stringify(arguments[arguments.length - 1]))
+        return returned;
+    };
+});
+
+module.exports = function (apiKey, metloHost) {
+    metloDetails.host = metloHost
+    metloDetails.key = apiKey
+}

ingestors/express/src/pool/index.js

@@ -0,0 +1,88 @@
+const { AsyncResource } = require('node:async_hooks');
+const { EventEmitter } = require('node:events');
+const path = require('node:path');
+const { Worker } = require('node:worker_threads');
+
+const kTaskInfo = Symbol('kTaskInfo');
+const kWorkerFreedEvent = Symbol('kWorkerFreedEvent');
+
+class WorkerPoolTaskInfo extends AsyncResource {
+    constructor(callback) {
+        super('WorkerPoolTaskInfo');
+        this.callback = callback;
+    }
+
+    done(err, result) {
+        this.runInAsyncScope(this.callback, null, err, result);
+        this.emitDestroy();  // `TaskInfo`s are used only once.
+    }
+}
+
+class WorkerPool extends EventEmitter {
+    constructor(numThreads, fileName) {
+        super();
+        this.numThreads = numThreads;
+        this.workers = [];
+        this.freeWorkers = [];
+        this.tasks = [];
+        this.workerFileName = fileName
+
+        for (let i = 0; i < numThreads; i++)
+            this.addNewWorker();
+
+        // Any time the kWorkerFreedEvent is emitted, dispatch
+        // the next task pending in the queue, if any.
+        this.on(kWorkerFreedEvent, () => {
+            if (this.tasks.length > 0) {
+                const { task, callback } = this.tasks.shift();
+                this.runTask(task, callback);
+            }
+        });
+    }
+
+    addNewWorker() {
+        const worker = new Worker(path.resolve(__dirname, this.workerFileName));
+        worker.on('message', (result) => {
+            // In case of success: Call the callback that was passed to `runTask`,
+            // remove the `TaskInfo` associated with the Worker, and mark it as free
+            // again.
+            worker[kTaskInfo].done(null, result);
+            worker[kTaskInfo] = null;
+            this.freeWorkers.push(worker);
+            this.emit(kWorkerFreedEvent);
+        });
+        worker.on('error', (err) => {
+            // In case of an uncaught exception: Call the callback that was passed to
+            // `runTask` with the error.
+            if (worker[kTaskInfo])
+                worker[kTaskInfo].done(err, null);
+            else
+                this.emit('error', err);
+            // Remove the worker from the list and start a new Worker to replace the
+            // current one.
+            this.workers.splice(this.workers.indexOf(worker), 1);
+            this.addNewWorker();
+        });
+        this.workers.push(worker);
+        this.freeWorkers.push(worker);
+        this.emit(kWorkerFreedEvent);
+    }
+
+    runTask(task, callback) {
+        if (this.freeWorkers.length === 0) {
+            // No free threads, wait until a worker thread becomes free.
+            this.tasks.push({ task, callback });
+            return;
+        }
+
+        const worker = this.freeWorkers.pop();
+        worker[kTaskInfo] = new WorkerPoolTaskInfo(callback);
+        worker.postMessage(task);
+    }
+
+    close() {
+        for (const worker of this.workers) worker.terminate();
+    }
+}
+
+module.exports = WorkerPool;

ingestors/express/src/pool/workerTarget.js

@@ -0,0 +1,31 @@
+const { parentPort } = require('node:worker_threads');
+const https = require('https')
+parentPort.on('message', ({ data: postData, host, key }) => {
+    console.log(postData)
+    var _resp = https.request(host, {
+        method: "POST", headers: {
+            'Content-Type': 'application/json',
+            'Content-Length': Buffer.byteLength(postData),
+            'Authorization': key
+        }
+    }, (res) => {
+        let data = '';
+
+        console.log('Status Code:', res.statusCode);
+
+        res.on('data', (chunk) => {
+            data += chunk;
+        });
+
+        res.on('end', () => {
+            console.log('Body: ', JSON.parse(data));
+        });
+
+    }).on("error", (err) => {
+        console.log("Error: ", err.message);
+    })
+
+    _resp.write(postData)
+    _resp.end();
+    parentPort.postMessage("Sent Message");
+});