[lambda][alert] adding support for sns as a source

Updating tests to include sns sources.

conf/sources.json

@@ -22,5 +22,16 @@
 				"cloudtrail"
 			]
 		}
+	},
+	"sns": {
+		"example_sns_topic": {
+			"logs": [
+				"json_log",
+				"syslog_log",
+				"kv_log",
+				"csv_log",
+				"osquery"
+			]
+		}
 	}
 }

stream_alert/rule_processor/classifier.py

@@ -130,11 +130,14 @@ def map_source(self, payload):
             payload.service = 'kinesis'
         elif 's3' in payload.raw_record:
             payload.service = 's3'
+        elif 'Sns' in payload.raw_record:
+            payload.service = 'sns'
 
         # map the entity name from a record
         entity_mapper = {
             'kinesis': lambda r: r['eventSourceARN'].split('/')[1],
-            's3': lambda r: r['s3']['bucket']['name']
+            's3': lambda r: r['s3']['bucket']['name'],
+            'sns': lambda r: r['EventSubscriptionArn'].split(':')[5]
         }
         # get the entity name
         payload.entity = entity_mapper[payload.service](payload.raw_record)

stream_alert/rule_processor/config.py

@@ -67,8 +67,8 @@ def validate_config(config):
 
         # check sources attributes
         elif config_key == 'sources':
-            if not set(settings.keys()).issubset(set(['kinesis', 's3'])):
-                raise ConfigError('Sources missing kinesis or s3 keys')
+            if not set(settings.keys()).issubset(set(['kinesis', 's3', 'sns'])):
+                raise ConfigError('Sources missing kinesis, s3 or sns keys')
             for log, attrs in settings.iteritems():
                 for entity, entity_attrs in attrs.iteritems():
                     if 'logs' not in set(entity_attrs.keys()):

stream_alert/rule_processor/handler.py

@@ -62,6 +62,8 @@ def run(self, event, context):
                 self.s3_process(payload, classifier)
             elif payload.service == 'kinesis':
                 self.kinesis_process(payload, classifier)
+            elif payload.service == 'sns':
+                self.sns_process(payload, classifier)
             else:
                 logger.info('Unsupported service: %s', payload.service)
 
@@ -86,6 +88,12 @@ def s3_process(self, payload, classifier):
             classifier.classify_record(payload, data)
             self.process_alerts(payload)
 
+    def sns_process(self, payload, classifier):
+        """Process SNS data for alerts"""
+        data = StreamPreParsers.pre_parse_sns(payload.raw_record)
+        classifier.classify_record(payload, data)
+        self.process_alerts(payload)
+
     def send_alerts(self, env, payload):
         """Send generated alerts to correct places"""
         if self.alerts:

stream_alert/rule_processor/pre_parsers.py

@@ -67,6 +67,17 @@ def pre_parse_s3(cls, raw_record):
 
         return cls._read_s3_file(downloaded_s3_object)
 
+    @classmethod
+    def pre_parse_sns(cls, raw_record):
+        """Decode an SNS record.
+
+        Args:
+            raw_record (dict): An SNS message.
+
+        Returns: (string) Base64 decoded data.
+        """
+        return base64.b64decode(raw_record['Sns']['Message'])
+
     @classmethod
     def _download_s3_object(cls, client, bucket, key, size):
         """Download an object from S3.

stream_alert_cli/test.py

@@ -140,9 +140,8 @@ def format_record(test_record):
         template['eventSourceARN'] = 'arn:aws:kinesis:us-east-1:111222333:stream/{}'.format(source)
 
     elif service == 'sns':
-        # TODO implement sns testing
-        raise NotImplementedError
-
+        template['Sns']['Message'] = base64.b64encode(data)
+        template['EventSubscriptionArn'] = 'arn:aws:sns:us-east-1:111222333:{}'.format(source)
     else:
         LOGGER_CLI.info('Invalid service %s', service)
 
@@ -244,8 +243,7 @@ def test_alert_rules():
                 if not check_keys(test_record):
                     report_output([test_record['service'],
                                    'Improperly formatted record: {}'.format(test_record)],
-                                  True
-                                 )
+                                  True)
                     tests_passed = False
                     continue
 

test/integration/rules/invalid_subnet.json

@@ -15,7 +15,7 @@
           "environment": "qa"
         }
       },
-      "description": "user logging in from an untrusted subnet",
+      "description": "user logging in from an untrusted subnet (10.4.0.16)",
       "trigger": true,
       "source": "prefix_cluster1_stream_alert_kinesis",
       "service": "kinesis"
@@ -35,7 +35,7 @@
           "environment": "qa"
         }
       },
-      "description": "user logging in from the trusted subnet",
+      "description": "user logging in from the trusted subnet (10.2.0.16)",
       "trigger": false,
       "source": "prefix_cluster1_stream_alert_kinesis",
       "service": "kinesis"
@@ -55,7 +55,7 @@
           "environment": "qa"
         }
       },
-      "description": "user logging in from an untrusted subnet",
+      "description": "user logging in from an untrusted subnet (10.54.0.21)",
       "trigger": true,
       "source": "example_s3_bucket",
       "service": "s3"
@@ -75,10 +75,50 @@
           "environment": "qa"
         }
       },
-      "description": "user logging in from the trusted subnet",
+      "description": "user logging in from the trusted subnet (10.2.0.99)",
       "trigger": false,
       "source": "example_s3_bucket",
       "service": "s3"
+    },
+    {
+      "data": {
+        "name": "logged_in_users",
+        "hostIdentifier": "host3",
+        "calendarTime": "Tue Feb 24 12:20:48 UTC 2017",
+        "unixTime": "1488309540",
+        "columns": {
+          "host": "10.99.0.210",
+          "user": "tom"
+        },
+        "action": "added",
+        "decorations": {
+          "environment": "qa"
+        }
+      },
+      "description": "user logging in from an untrusted subnet (10.99.0.210)",
+      "trigger": true,
+      "source": "example_sns_topic",
+      "service": "sns"
+    },
+    {
+      "data": {
+        "name": "logged_in_users",
+        "hostIdentifier": "host4",
+        "calendarTime": "Tue Feb 21 20:16:55 UTC 2017",
+        "unixTime": "1488309540",
+        "columns": {
+          "host": "10.2.0.244",
+          "user": "charlie"
+        },
+        "action": "added",
+        "decorations": {
+          "environment": "qa"
+        }
+      },
+      "description": "user logging in from the trusted subnet (10.2.0.244)",
+      "trigger": false,
+      "source": "example_sns_topic",
+      "service": "sns"
     }
   ]
 }

test/integration/rules/invalid_user.json

@@ -15,7 +15,7 @@
           "environment": "qa"
         }
       },
-      "description": "user not in the whitelist",
+      "description": "user not in the whitelist (john)",
       "trigger": true,
       "source": "prefix_cluster1_stream_alert_kinesis",
       "service": "kinesis"
@@ -35,7 +35,7 @@
           "environment": "qa"
         }
       },
-      "description": "user in the whitelist",
+      "description": "user in the whitelist (alice)",
       "trigger": false,
       "source": "prefix_cluster1_stream_alert_kinesis",
       "service": "kinesis"
@@ -55,7 +55,7 @@
           "environment": "qa"
         }
       },
-      "description": "user not in the whitelist",
+      "description": "user not in the whitelist (sam)",
       "trigger": true,
       "source": "example_s3_bucket",
       "service": "s3"
@@ -75,10 +75,50 @@
           "environment": "qa"
         }
       },
-      "description": "user in the whitelist",
+      "description": "user in the whitelist (bob)",
       "trigger": false,
       "source": "example_s3_bucket",
       "service": "s3"
+    },
+    {
+      "data": {
+        "name": "logged_in_users",
+        "hostIdentifier": "host5",
+        "calendarTime": "Tue Feb 12 10:12:55 UTC 2017",
+        "unixTime": "1488309540",
+        "columns": {
+          "host": "10.2.0.16",
+          "user": "alice"
+        },
+        "action": "added",
+        "decorations": {
+          "environment": "qa"
+        }
+      },
+      "description": "user not in the whitelist (alice)",
+      "trigger": false,
+      "source": "example_sns_topic",
+      "service": "sns"
+    },
+    {
+      "data": {
+        "name": "logged_in_users",
+        "hostIdentifier": "host6",
+        "calendarTime": "Tue Feb 11 11:11:11 UTC 2017",
+        "unixTime": "1488309540",
+        "columns": {
+          "host": "10.2.0.16",
+          "user": "tom"
+        },
+        "action": "added",
+        "decorations": {
+          "environment": "qa"
+        }
+      },
+      "description": "user in the whitelist (tom)",
+      "trigger": true,
+      "source": "example_sns_topic",
+      "service": "sns"
     }
   ]
 }

test/integration/rules/sample_csv_rule.json

@@ -13,6 +13,13 @@
       "trigger": true,
       "source": "example_s3_bucket",
       "service": "s3"
+    },
+    {
+      "data": "Jan 01 2017,1487095529,test-host-2,this is test data for rules,cluster 6",
+      "description": "host is test-host-2",
+      "trigger": true,
+      "source": "example_sns_topic",
+      "service": "sns"
     }
   ]
 }

test/integration/rules/sample_json_rule.json

@@ -25,6 +25,19 @@
       "trigger": true,
       "source": "example_s3_bucket",
       "service": "s3"
+    },
+    {
+      "data": {
+        "name": "name-1",
+        "host": "test-host-1",
+        "data": {
+          "time": "Jan 01, 2017"
+        }
+      },
+      "description": "host is test-host-1",
+      "trigger": true,
+      "source": "example_sns_topic",
+      "service": "sns"
     }
   ]
 }

test/integration/rules/sample_kv_rule.json

@@ -13,6 +13,13 @@
       "trigger": true,
       "source": "example_s3_bucket",
       "service": "s3"
+    },
+    {
+      "data": "type=test msg=fatal uid=100 time=1488013995",
+      "description": "fatal message from uid 100",
+      "trigger": true,
+      "source": "example_sns_topic",
+      "service": "sns"
     }
   ]
 }

test/integration/rules/sample_kv_rule_last_hour.json

@@ -13,6 +13,13 @@
       "trigger": true,
       "source": "example_s3_bucket",
       "service": "s3"
+    },
+    {
+      "data": "type=start msg=info uid=0 time=<helper:last_hour>",
+      "description": "info message from uid 0 in the last hour",
+      "trigger": true,
+      "source": "example_sns_topic",
+      "service": "sns"
     }
   ]
 }

test/integration/rules/sample_syslog_rule.json

@@ -2,17 +2,24 @@
   "records": [
     {
       "data": "Jan 01 12:00:12 test-host-1 sudo[151]: COMMAND sudo rm /tmp/test",
-      "description": "sudo command ran",
+      "description": "sudo command ran (`sudo rm /tmp/test`)",
       "trigger": false,
       "source": "prefix_cluster1_stream_alert_kinesis",
       "service": "kinesis"
     },
     {
       "data": "Jan 01 12:00:12 test-host-2 sudo[159]: COMMAND sudo rm /tmp/badstuff",
-      "description": "sudo command ran",
+      "description": "sudo command ran (`sudo rm /tmp/badstuff`)",
       "trigger": false,
       "source": "example_s3_bucket",
       "service": "s3"
+    },
+    {
+      "data": "Jan 01 12:00:12 test-host-2 sudo[159]: COMMAND sudo rm -rf /",
+      "description": "sudo command ran (`sudo rm -rf /`)",
+      "trigger": false,
+      "source": "example_sns_topic",
+      "service": "sns"
     }
   ]
 }