Merge pull request #123 from SumoLogic/hpal_SUMO-252521

Added support for Virtual Network Flow Logs

AppendBlobReader/README.md

@@ -35,16 +35,17 @@ This command copies required files in `AppendBlobReader/target` directory
 
 Integrations tests are in `AppendBlobReader/tests` folder and unit tests are in sumo-`function-utils/tests` folder
 
+Modify the run_integration_test.sh file with below parameters
 ```console
 
-export AZURE_SUBSCRIPTION_ID=`<Your azure subscription id, to obtain it refer docs https://learn.microsoft.com/en-us/azure/azure-portal/get-subscription-tenant-id#find-your-azure-subscription>`
-export AZURE_CLIENT_ID=`Your application id which you can get after registering application. Refer https://learn.microsoft.com/en-us/entra/identity-platform/quickstart-register-app#register-an-application`
-export AZURE_CLIENT_SECRET=`Generate client secret by referring docs https://learn.microsoft.com/en-us/entra/identity-platform/quickstart-register-app#add-credentials`
-export AZURE_TENANT_ID=`You tenant id, to obtain it refer docs https://learn.microsoft.com/en-us/azure/azure-portal/get-subscription-tenant-id#find-your-microsoft-entra-tenant`
-export AZURE_DEFAULT_REGION=`eastus`
-export SUMO_ACCESS_ID=`<Generate access id and access key https://help.sumologic.com/docs/manage/security/access-keys/#create-your-access-key>`
-export SUMO_ACCESS_KEY=`<Generate access id and access key https://help.sumologic.com/docs/manage/security/access-keys/#create-your-access-key>`
-export SUMO_DEPLOYMENT=`Enter one of the allowed values au, ca, de, eu, fed, in, jp, us1 or us2. Visit https://help.sumologic.com/APIs/General-API-Information/Sumo-Logic-Endpoints-and-Firewall-Security`
+AZURE_SUBSCRIPTION_ID=`<Your azure subscription id, to obtain it refer docs https://learn.microsoft.com/en-us/azure/azure-portal/get-subscription-tenant-id#find-your-azure-subscription>`
+AZURE_CLIENT_ID=`Your application id which you can get after registering application. Refer https://learn.microsoft.com/en-us/entra/identity-platform/quickstart-register-app#register-an-application`
+AZURE_CLIENT_SECRET=`Generate client secret by referring docs https://learn.microsoft.com/en-us/entra/identity-platform/quickstart-register-app#add-credentials`
+AZURE_TENANT_ID=`You tenant id, to obtain it refer docs https://learn.microsoft.com/en-us/azure/azure-portal/get-subscription-tenant-id#find-your-microsoft-entra-tenant`
+AZURE_DEFAULT_REGION=`eastus`
+SUMO_ACCESS_ID=`<Generate access id and access key https://help.sumologic.com/docs/manage/security/access-keys/#create-your-access-key>`
+SUMO_ACCESS_KEY=`<Generate access id and access key https://help.sumologic.com/docs/manage/security/access-keys/#create-your-access-key>`
+SUMO_DEPLOYMENT=`Enter one of the allowed values au, ca, de, eu, fed, in, jp, us1 or us2. Visit https://help.sumologic.com/APIs/General-API-Information/Sumo-Logic-Endpoints-and-Firewall-Security`
 ```
 
 Execute below command under `AppendBlobReader/tests` directory

AppendBlobReader/tests/run_integration_test.sh

@@ -0,0 +1,15 @@
+#!/bin/bash
+
+export AZURE_SUBSCRIPTION_ID=""
+# application id
+export AZURE_CLIENT_ID=""
+export AZURE_CLIENT_SECRET=""
+export AZURE_TENANT_ID=""
+export AZURE_DEFAULT_REGION="eastus"
+export SUMO_ACCESS_ID=""
+export SUMO_ACCESS_KEY=""
+export SUMO_DEPLOYMENT="us1"
+export TEMPLATE_NAME="appendblobreaderdeploy.json"
+python test_appendblobreader.py
+# For deleting leftover resources in case of failures
+# python ~/git/sumologic-azure-function/deletetestresourcegroups.py

BlockBlobReader/README.md

@@ -37,19 +37,21 @@ Integrations tests are in `BlockBlobReader/tests` folder and unit tests are in `
 
 Integrations tests are in `BlockBlobReader/tests` folder and unit tests are in sumo-`function-utils/tests` folder
 
-```console
+Modify the run_integration_test.sh file with below parameters
 
-export AZURE_SUBSCRIPTION_ID=`<Your azure subscription id, to obtain it refer docs https://learn.microsoft.com/en-us/azure/azure-portal/get-subscription-tenant-id#find-your-azure-subscription>`
-export AZURE_CLIENT_ID=`Your application id which you can get after registering application. Refer https://learn.microsoft.com/en-us/entra/identity-platform/quickstart-register-app#register-an-application`
-export AZURE_CLIENT_SECRET=`Generate client secret by referring docs https://learn.microsoft.com/en-us/entra/identity-platform/quickstart-register-app#add-credentials`
-export AZURE_TENANT_ID=`You tenant id, to obtain it refer docs https://learn.microsoft.com/en-us/azure/azure-portal/get-subscription-tenant-id#find-your-microsoft-entra-tenant`
-export AZURE_DEFAULT_REGION=`eastus`
-export SUMO_ACCESS_ID=`<Generate access id and access key https://help.sumologic.com/docs/manage/security/access-keys/#create-your-access-key>`
-export SUMO_ACCESS_KEY=`<Generate access id and access key https://help.sumologic.com/docs/manage/security/access-keys/#create-your-access-key>`
-export SUMO_DEPLOYMENT=`Enter one of the allowed values au, ca, de, eu, fed, in, jp, us1 or us2. Visit https://help.sumologic.com/APIs/General-API-Information/Sumo-Logic-Endpoints-and-Firewall-Security`
+```console
 
+AZURE_SUBSCRIPTION_ID=`<Your azure subscription id, to obtain it refer docs https://learn.microsoft.com/en-us/azure/azure-portal/get-subscription-tenant-id#find-your-azure-subscription>`
+AZURE_CLIENT_ID=`Your application id which you can get after registering application. Refer https://learn.microsoft.com/en-us/entra/identity-platform/quickstart-register-app#register-an-application`
+AZURE_CLIENT_SECRET=`Generate client secret by referring docs https://learn.microsoft.com/en-us/entra/identity-platform/quickstart-register-app#add-credentials`
+AZURE_TENANT_ID=`You tenant id, to obtain it refer docs https://learn.microsoft.com/en-us/azure/azure-portal/get-subscription-tenant-id#find-your-microsoft-entra-tenant`
+AZURE_DEFAULT_REGION=`eastus`
+SUMO_ACCESS_ID=`<Generate access id and access key https://help.sumologic.com/docs/manage/security/access-keys/#create-your-access-key>`
+SUMO_ACCESS_KEY=`<Generate access id and access key https://help.sumologic.com/docs/manage/security/access-keys/#create-your-access-key>`
+SUMO_DEPLOYMENT=`Enter one of the allowed values au, ca, de, eu, fed, in, jp, us1 or us2. Visit https://help.sumologic.com/APIs/General-API-Information/Sumo-Logic-Endpoints-and-Firewall-Security`
 ```
 
+
 Execute below command under `BlockBlobReader/tests` directory
 
 `python test_blobreader.py`

BlockBlobReader/src/blobreaderzipdeploy.json

@@ -419,7 +419,7 @@
                       "[variables('BlobReader_resourceId')]"
                   ],
                   "properties": {
-                    "packageUri": "https://appdev-cloudformation-templates.s3.amazonaws.com/AzureBlobReader/taskproducer4.1.4.zip",
+                    "packageUri": "https://appdev-cloudformation-templates.s3.amazonaws.com/AzureBlobReader/taskproducer4.1.5.zip",
                     "appOffline": true
                   }
                 }
@@ -512,7 +512,7 @@
                       "[variables('blobreaderconsumer_resourceId')]"
                   ],
                   "properties": {
-                    "packageUri": "https://appdev-cloudformation-templates.s3.amazonaws.com/AzureBlobReader/taskconsumer4.1.4.zip",
+                    "packageUri": "https://appdev-cloudformation-templates.s3.amazonaws.com/AzureBlobReader/taskconsumer4.1.5.zip",
                     "appOffline": true
                   }
                 }
@@ -609,7 +609,7 @@
                       "[variables('DLQProcessor_resourceId')]"
                   ],
                   "properties": {
-                    "packageUri": "https://appdev-cloudformation-templates.s3.amazonaws.com/AzureBlobReader/dlqprocessor4.1.4.zip",
+                    "packageUri": "https://appdev-cloudformation-templates.s3.amazonaws.com/AzureBlobReader/dlqprocessor4.1.5.zip",
                     "appOffline": true
                   }
                 }

BlockBlobReader/src/consumer.js

@@ -218,7 +218,7 @@ async function setAppendBlobOffset(context, serviceBusTask, newOffset) {
 
 async function nsgLogsHandler(context, msg, serviceBusTask) {
 
-    var jsonArray = [];
+    let jsonArray = [];
     msg = msg.trim().replace(/(^,)|(,$)/g, ""); //removing trailing spaces,newlines and leftover commas
 
     try {
@@ -236,7 +236,7 @@ async function nsgLogsHandler(context, msg, serviceBusTask) {
 
     }
 
-    var eventsArr = [];
+    let eventsArr = [];
     jsonArray.forEach(function (record) {
         let version = record.properties.Version;
         record.properties.flows.forEach(function (rule) {
@@ -282,6 +282,66 @@ async function nsgLogsHandler(context, msg, serviceBusTask) {
     return eventsArr;
 }
 
+async function networkFlowLogsHandler(context, msg, serviceBusTask) {
+
+    let jsonArray = [];
+    msg = msg.trim().replace(/(^,)|(,$)/g, ""); //removing trailing spaces,newlines and leftover commas
+
+    try {
+        jsonArray = JSON.parse("[" + msg + "]");
+    } catch(err) {
+        let response = getParseableJsonArray(msg, context, serviceBusTask);
+        jsonArray = response[0];
+        let is_success = response[2];
+        let newOffset = response[1] + serviceBusTask.startByte;
+        if (is_success) {
+            await setAppendBlobOffset(context, serviceBusTask, newOffset);
+        } else {
+            return jsonArray;
+        }
+
+    }
+    // Format: https://learn.microsoft.com/en-us/azure/network-watcher/vnet-flow-logs-overview#log-format
+    let eventsArr = [];
+    jsonArray.forEach(function (record) {
+        record.flowRecords.flows.forEach(function (acl) {
+            acl.flowGroups.forEach(function (rule) {
+                rule.flowTuples.forEach(function (tuple) {
+                    let col = tuple.split(",");
+                    let event = {
+                        time: col[0], // Time stamp of when the flow occurred, in UNIX epoch format.
+                        flowLogGUID: record.flowLogGUID,
+                        category: record.category,
+                        flow_log_resource_id: record.flowLogResourceID,
+                        target_resource_id: record.targetResourceID,
+                        event_name: record.operationName,
+                        acl_id: acl.aclID,
+                        rule_name: rule.rule,
+                        mac: record.macAddress,
+                        src_ip: col[1],
+                        dest_IP: col[2],
+                        src_port: col[3],
+                        dest_port: col[4],
+                        protocol: col[5],
+                        flow_direction: col[6],
+                        flow_state: (col[7] === "" || col[7] === undefined) ?  null : col[7],
+                        flow_encryption_status: (col[8] === "" || col[8] === undefined) ?  null : col[8],
+                        num_packets_sent_src_to_dest: (col[9] === "" || col[9] === undefined) ?  null : col[9],
+                        bytes_sent_src_to_dest: (col[10] === "" || col[10] === undefined) ?  null : col[10],
+                        num_packets_sent_dest_to_src: (col[11] === "" || col[11] === undefined) ?  null : col[11],
+                        bytes_sent_dest_to_src: (col[12] === "" || col[12] === undefined) ?  null : col[12],
+                        version: record.flowLogVersion
+
+                    }
+                    eventsArr.push(event);
+                });
+            });
+        });
+    });
+    return eventsArr;
+}
+
+
 function jsonHandler(context,msg) {
     // it's assumed that json is well formed {},{}
     var jsonArray = [];
@@ -347,13 +407,13 @@ function messageHandler(serviceBusTask, context, sumoClient) {
     if (file_ext == serviceBusTask.blobName) {
         file_ext = "log";
     }
-    var msghandler = {"log": logHandler, "csv": csvHandler, "json": jsonHandler, "blob": blobHandler, "nsg": nsgLogsHandler};
+    var msghandler = {"log": logHandler, "csv": csvHandler, "json": jsonHandler, "blob": blobHandler, "nsg": nsgLogsHandler, "vnetflowlogs": networkFlowLogsHandler};
     if (!(file_ext in msghandler)) {
         context.log.error("Error in messageHandler: Unknown file extension - " + file_ext + " for blob: " + serviceBusTask.blobName);
         context.done();
         return;
     }
-    if ((file_ext === "json") && (serviceBusTask.containerName === "insights-logs-networksecuritygroupflowevent")) {
+    if ((file_ext === "json") && (serviceBusTask.containerName === "insights-logs-networksecuritygroupflowevent" || serviceBusTask.containerName === "insights-logs-flowlogflowevent")) {
         // because in json first block and last block remain as it is and azure service adds new block in 2nd last pos
         if ((serviceBusTask.endByte < JSON_BLOB_HEAD_BYTES + JSON_BLOB_TAIL_BYTES) || (serviceBusTask.endByte == serviceBusTask.startByte)) {
             context.done(); //rejecting first commit when no data is there data will always be atleast HEAD_BYTES+DATA_BYTES+TAIL_BYTES
@@ -365,7 +425,7 @@ function messageHandler(serviceBusTask, context, sumoClient) {
         } else {
             serviceBusTask.startByte -= 1; //to remove comma before json object
         }
-        file_ext = "nsg";
+        file_ext = serviceBusTask.containerName === "insights-logs-networksecuritygroupflowevent" ? "nsg" : "vnetflowlogs";
     }
     getBlockBlobService(context, serviceBusTask).then(function (blobService) {
         return getData(serviceBusTask, blobService, context).then(async function (msg) {
@@ -385,10 +445,10 @@ function messageHandler(serviceBusTask, context, sumoClient) {
                     context.done(err);
                 });
             } else {
-                if (file_ext == "nsg") {
-                    messageArray = await nsgLogsHandler(context, msg, serviceBusTask);
+                if (file_ext === "nsg" || file_ext === "vnetflowlogs") {
+                    messageArray = await msghandler[file_ext](context, msg, serviceBusTask);
                 } else {
-                    messageArray = msghandler[file_ext](context,msg);
+                    messageArray = msghandler[file_ext](context, msg);
                 }
                 messageArray.forEach(function (msg) {
                     sumoClient.addData(msg);

BlockBlobReader/src/create_zip.sh

@@ -80,7 +80,7 @@ else
 fi
 
 echo "creating zip"
-version="4.1.4"
+version="4.1.5"
 producer_zip_file="taskproducer$version.zip"
 consumer_zip_file="taskconsumer$version.zip"
 dlqprocessor_zip_file="dlqprocessor$version.zip"