Update aws log forwarder

impart-security · Jan 9, 2025 · 19d9a46 · 19d9a46
1 parent 1ea1ec8
commit 19d9a46
Show file tree

Hide file tree

Showing 9 changed files with 164 additions and 1,578 deletions.
diff --git a/.eslintrc.json b/.eslintrc.json
diff --git a/.github/workflows/release.yaml b/.github/workflows/release.yaml
@@ -1,3 +1,4 @@
+---
 name: release
 on:
   push:
@@ -14,9 +15,10 @@ jobs:
         uses: actions/setup-node@v4
         with:
           node-version: 20
+          cache: npm
       - name: Package
         run: |
-          npm ci
+          npm i
           zip -r aws-log-forwarder.zip  . -i "*.js" "*.json" "node_modules" VERSION LICENSE
       - name: Release
         uses: softprops/action-gh-release@v2

diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,5 +1,12 @@
+
 # Changelog
 
+## [0.2.1] - 2024-01-09
+
+### Changed
+
+- Updated dependencies.
+
 ## [0.2.0] - 2024-04-03
 
 ### Added
@@ -11,3 +18,4 @@
 ### Added
 
 - Initial release
+
diff --git a/VERSION b/VERSION
@@ -1 +1 @@
-0.2.0
+0.2.1
diff --git a/eslint.config.mjs b/eslint.config.mjs
@@ -0,0 +1,9 @@
+import eslint from "@eslint/js";
+import globals from "globals";
+
+export default {
+  ...eslint.configs.recommended,
+  languageOptions: {
+    globals: globals.node,
+  },
+};
diff --git a/index.js b/index.js
@@ -1,15 +1,24 @@
 "use strict";
 
+// @ts-check
+
 const { GetObjectCommand, S3Client } = require("@aws-sdk/client-s3");
 const zlib = require("zlib");
 const util = require("util");
 const gunzip = util.promisify(zlib.gunzip);
 const readline = require("readline");
 const dgram = require("dgram");
+const stream = require("stream");
 
-const inspectorAddr = process.env.INSPECTOR_LOGSTREAM_LISTEN_ADDR;
+const inspectorAddr = process.env["INSPECTOR_LOGSTREAM_LISTEN_ADDR"];
 
-exports.handler = async (event, context, callback) => {
+/**
+ * AWS Lambda handler
+ * @param { {awslogs?: {data: string}, Records: {s3?: {bucket: {name:string}, object: {key: string}}}[] } } event - The event object
+ * @param {Object} _context - The context object
+ * @param {Function} callback - The callback function
+ */
+exports.handler = async (event, _context, callback) => {
   if (!inspectorAddr) {
     const err = "missing INSPECTOR_LOGSTREAM_LISTEN_ADDR env variable";
     console.log(err);
@@ -25,59 +34,67 @@ exports.handler = async (event, context, callback) => {
     return;
   }
 
-  const [inspectorHost, inspectorPort] = arr;
+  const [inspectorHost, inspectorPortStr] = arr;
+  const inspectorPort = inspectorPortStr ? parseInt(inspectorPortStr, 10) : 0;
+  if (isNaN(inspectorPort)) {
+    callback(`invalid inspector port: ${inspectorPortStr}`);
+    return;
+  }
+
   const client = dgram.createSocket("udp4");
 
+  const record = event.Records[0];
+
   if (event.awslogs) {
     console.log("awslogs event");
-    const payload = new Buffer.from(event.awslogs.data, "base64"); // decode base64 to binary
+    const payload = Buffer.from(event.awslogs.data, "base64"); // decode base64 to binary
     const result = await gunzip(payload);
     const parsedRequest = JSON.parse(result.toString("utf8"));
 
-    await new Promise((resolve) => {
-      for (let i = 0; i < parsedRequest.logEvents.length; i++) {
-        if (
-          parsedRequest.logEvents[i].message.length &&
-          parsedRequest.logEvents[i].message[0] === "#"
-        ) {
-          continue;
-        }
+    await /** @type {Promise<void>} */ (
+      new Promise((resolve) => {
+        for (let i = 0; i < parsedRequest.logEvents.length; i++) {
+          if (
+            parsedRequest.logEvents[i].message.length &&
+            parsedRequest.logEvents[i].message[0] === "#"
+          ) {
+            continue;
+          }
+
+          const current = i;
+
+          const message = parsedRequest.logEvents[i].message.endsWith("\n")
+            ? Buffer.from(parsedRequest.logEvents[i].message)
+            : Buffer.from(parsedRequest.logEvents[i].message + "\n");
+          client.send(
+            message,
+            0,
+            message.length,
+            inspectorPort,
+            inspectorHost,
+            (err) => {
+              if (err) console.error(err);
 
-        const current = i;
-
-        const message = parsedRequest.logEvents[i].message.endsWith("\n")
-          ? Buffer.from(parsedRequest.logEvents[i].message)
-          : Buffer.from(parsedRequest.logEvents[i].message + "\n");
-        client.send(
-          message,
-          0,
-          message.length,
-          inspectorPort,
-          inspectorHost,
-          (err) => {
-            if (err) console.error(err);
-
-            if (current === parsedRequest.logEvents.length - 1) {
-              client.close();
-              console.log(
-                `sent ${parsedRequest.logEvents.length} lines for inspection`,
-              );
-              callback(
-                null,
-                `sent ${parsedRequest.logEvents.length} lines for inspection`,
-              );
-              resolve();
-            }
-          },
-        );
-      }
-    });
-  } else if (event.Records[0].s3) {
-    const bucket = event.Records[0].s3.bucket.name;
-    console.log("S3 bucket: ", bucket);
-    const key = decodeURIComponent(
-      event.Records[0].s3.object.key.replace(/\+/g, " "),
+              if (current === parsedRequest.logEvents.length - 1) {
+                client.close();
+                console.log(
+                  `sent ${parsedRequest.logEvents.length} lines for inspection`,
+                );
+                callback(
+                  null,
+                  `sent ${parsedRequest.logEvents.length} lines for inspection`,
+                );
+                resolve();
+              }
+            },
+          );
+        }
+      })
     );
+  } else if (!!record && !!record.s3) {
+    const bucket = record.s3.bucket.name;
+    console.log("S3 bucket: ", bucket);
+    const key = decodeURIComponent(record.s3.object.key.replace(/\+/g, " "));
 
     // Retrieve S3 Object
     const s3Client = new S3Client();
@@ -87,47 +104,70 @@ exports.handler = async (event, context, callback) => {
     });
 
     const response = await s3Client.send(getObjectCommand);
+    if (!response.Body) {
+      callback("no body in S3 object");
+      return;
+    }
+
+    /** @type {stream.Readable} */
+    let body;
+
+    // Check if body is a Blob and convert it to ReadableStream
+    if (response.Body instanceof Blob) {
+      const readableStream = response.Body.stream();
+      body = stream.Readable.from(readableStream);
+    } else if (response.Body instanceof stream.Readable) {
+      body = response.Body;
+    } else {
+      callback(
+        "Unexpected body type: response.Body is not a compatible stream type.",
+      );
+      return;
+    }
+
     const lineReader = readline.createInterface({
-      input: response.Body.pipe(zlib.createGunzip()),
+      input: body.pipe(zlib.createGunzip()),
     });
 
     let lineCount = 0;
     let sentCount = 0;
     let last = false;
 
-    await new Promise((resolve) => {
-      lineReader.on("line", (line) => {
-        if (line[0] !== "#") {
-          const message = Buffer.from(line + "\n");
-
-          ++lineCount;
-          client.send(
-            message,
-            0,
-            message.length,
-            inspectorPort,
-            inspectorHost,
-            (err) => {
-              if (err) console.error(err);
-
-              ++sentCount;
-
-              if (last && lineCount === sentCount) {
-                client.close();
-                console.log(`sent ${sentCount} lines for inspection`);
-                callback(null, `sent ${sentCount} lines for inspection`);
-                resolve();
-              }
-            },
-          );
-        }
-      });
-
-      lineReader.on("close", () => {
-        last = true;
-        console.log(`processed lines ${lineCount}`);
-      });
-    });
+    await /** @type {Promise<void>} */ (
+      new Promise((resolve) => {
+        lineReader.on("line", (line) => {
+          if (line[0] !== "#") {
+            const message = Buffer.from(line + "\n");
+
+            ++lineCount;
+            client.send(
+              message,
+              0,
+              message.length,
+              inspectorPort,
+              inspectorHost,
+              (err) => {
+                if (err) console.error(err);
+
+                ++sentCount;
+
+                if (last && lineCount === sentCount) {
+                  client.close();
+                  console.log(`sent ${sentCount} lines for inspection`);
+                  callback(null, `sent ${sentCount} lines for inspection`);
+                  resolve();
+                }
+              },
+            );
+          }
+        });
+
+        lineReader.on("close", () => {
+          last = true;
+          console.log(`processed lines ${lineCount}`);
+        });
+      })
+    );
   } else {
     callback("unsupported even type");
   }