PP-14422 Add AWS WAF Transformation

spec/fixtures/aws_waf_log_fixtures.ts

@@ -0,0 +1,47 @@
+import { Fixture } from './general_fixtures'
+
+export const anWAFCloudWatchEvent: Fixture = {
+  input: {
+    deliveryStreamArn: 'someDeliveryStreamArn',
+    invocationId: 'someId',
+    region: 'eu-west-1',
+    records: [{
+      approximateArrivalTimestamp: 1234,
+      recordId: 'record-id-waf-logs',
+      data: Buffer.from(JSON.stringify({
+        owner: '223851549868',
+        logGroup: 'aws-waf-logs-test-12',
+        logStream: 'logStream',
+        subscriptionFilters: [],
+        messageType: 'DATA_MESSAGE',
+        logEvents: [
+          {
+            id: 'cloudwatch-log-message-id-1',
+            timestamp: 1740495533,
+            message: '{"timestamp":1756899436824,"some":"json"}'
+          }
+        ]
+      })).toString('base64')
+    }]
+  },
+  expected: {
+    records: [{
+      result: 'Ok',
+      recordId: 'record-id-waf-logs',
+      data: Buffer.from([
+        {
+          host: 'logStream',
+          source: 'waf',
+          sourcetype: 'generic_single_line',
+          index: 'pay_ingress',
+          event: '{"timestamp":1756899436824,"some":"json"}',
+          fields: {
+            account: 'test',
+            environment: 'test-12'
+          },
+          time: 1756899436824
+        }
+      ].map(x => JSON.stringify(x)).join('\n')).toString('base64')
+    }]
+  }
+}

spec/index.test.ts

@@ -63,6 +63,9 @@ import {
   aMultiLogVpcFlowLogAllFilteredCloudWatchEvent,
   aMultiLogVpcFlowLogCloudWatchEvent
 } from './fixtures/vpcflowlog_fixtures'
+import {
+  anWAFCloudWatchEvent
+} from './fixtures/aws_waf_log_fixtures'
 import { SplunkRecord } from '../src/types'
 
 process.env.ENVIRONMENT = 'test-12'
@@ -332,6 +335,17 @@ describe('Processing CloudWatchLogEvents', () => {
     })
   })
 
+  describe('From AWS WAF', () => {
+    test('should send the log correctly transformed', async () => {
+      const result = await handler(anWAFCloudWatchEvent.input, mockContext, mockCallback) as FirehoseTransformationResult
+
+      const expected = anWAFCloudWatchEvent.expected.records[0]
+      expect(result.records[0].result).toEqual(expected.result)
+      expect(result.records[0].recordId).toEqual(expected.recordId)
+      expect(Buffer.from(result.records[0].data as string, 'base64').toString()).toEqual(Buffer.from(expected.data as string, 'base64').toString())
+    })
+  })
+
   test('should drop CloudWatch logs which are not DATA_MESSAGE', async () => {
     const result = await handler(aCloudWatchEventWith([{ messageType: 'CONTROL' }]), mockContext, mockCallback) as FirehoseTransformationResult
     expect(result.records[0].result).toEqual('Dropped')

src/extractTime.ts

@@ -79,6 +79,16 @@ export function extractAuditLogTime(log: string): number | undefined {
   return Number(extractedTime[1])
 }
 
+export function extractWAFLogTime(log: string): number | undefined {
+  const regex = /"timestamp"\s*:\s*(.*?),/
+  const extractedTime = regexTimeFromLog(regex, log)
+  if (extractedTime === undefined) {
+    return undefined
+  }
+
+  return parseInt(extractedTime[1])
+}
+
 export function extractConcourseLogTime(log: string): number | undefined {
   const regex = /"timestamp"\s*:\s*"(.*?)"/
   const extractedTime = regexTimeFromLog(regex, log)
@@ -175,5 +185,7 @@ export function parseTimeFromLog(log: string, logType: CloudWatchLogTypes): numb
       return extractNginxKvLogTime(log)
     case CloudWatchLogTypes.cloudtrail:
       return extractCloudTrailLogTime(log)
+    case CloudWatchLogTypes.waf:
+      return extractWAFLogTime(log)
   }
 }

src/transformData.ts

@@ -180,6 +180,8 @@ function sourceTypeFromLogGroup(logType: CloudWatchLogTypes, msg: string): strin
       return 'aws:cloudtrail'
     case CloudWatchLogTypes['vpc-flow-logs']:
       return 'aws:cloudwatchlogs:vpcflow'
+    case CloudWatchLogTypes['waf']:
+      return 'generic_single_line'
   }
 }
 
@@ -193,6 +195,7 @@ function indexFromLogType(logType: CloudWatchLogTypes): string {
       return 'pay_host'
     case CloudWatchLogTypes['nginx-forward-proxy']:
     case CloudWatchLogTypes['nginx-reverse-proxy']:
+    case CloudWatchLogTypes['waf']:
       return 'pay_ingress'
     case CloudWatchLogTypes['apt']:
     case CloudWatchLogTypes['audit']:
@@ -212,9 +215,11 @@ function indexFromLogType(logType: CloudWatchLogTypes): string {
 }
 
 function validateLogGroup(logGroup: string): void {
-  if (logGroup.match(LOG_GROUP_REGEX) === null) {
-    throw new Error(`Log group "${logGroup}" must be of format <env>_<type>_<optional subtype> matching ${LOG_GROUP_REGEX.toString()}`)
+  if (logGroup.startsWith('aws-waf-logs-') || logGroup.match(LOG_GROUP_REGEX) !== null) {
+    return
   }
+
+  throw new Error(`Log group "${logGroup}" must be of format <env>_<type>_<optional subtype> matching ${LOG_GROUP_REGEX.toString()}`)
 }
 
 function squidSourceType(msg: string): string {
@@ -225,6 +230,10 @@ function squidSourceType(msg: string): string {
 }
 
 function getLogTypeFromLogGroup(logGroup: string): CloudWatchLogTypes {
+  if (logGroup.startsWith('aws-waf-logs')) {
+    return CloudWatchLogTypes.waf
+  }
+
   const logType: string = logGroup.split('_')[1]
 
   if (Object.values(CloudWatchLogTypes).includes(logType)) {

src/types.ts

@@ -52,5 +52,6 @@ export enum CloudWatchLogTypes {
   'nginx-reverse-proxy',
   'syslog',
   'squid',
-  'vpc-flow-logs'
+  'vpc-flow-logs',
+  'waf'
 }