airbnb/streamalert

StreamAlert is a serverless, realtime data analysis framework which empowers you to ingest, analyze, and alert on data from any environment, using datasources and alerting logic you define.

StreamAlert - Serverless, Realtime Data Analysis Framework

StreamAlert is a serverless, real-time data analysis framework which empowers you to ingest, analyze, and alert on data from any environment, using data sources and alerting logic you define. Computer security teams use StreamAlert to scan terabytes of log data every day for incident detection and response.

Features

• Rules are written in Python; they can utilize any Python libraries or functions
• Ingested logs and generated alerts can be retroactively searched for compliance and research
• Serverless design is cheaper, easier to maintain, and scales to terabytes per day
• Deployment is automated: simple, safe and repeatable for any AWS account
• Secure by design: least-privilege execution, containerized analysis, and encrypted data storage
• Merge similar alerts and automatically promote new rules if they are not too noisy
• Built-in support for dozens of log types and schemas
• Built-in collection of broadly applicable community rules
• Fully open source and customizable: add your own log schemas, rules, and alert outputs

Ready? Let's get started!

Resources

• Blog Post
• User Guide
• Slack (unofficial)

streamalert's Issues

Improvement: JSON Cluster Templates

Background

The $ ./stream_alert_cli.py terraform generate command is used to dynamically create Terraform files which represent a StreamAlert cluster. This works by using a cluster_template, and passing in variables to render a desired cluster plan. Since Terraform plans can either be HCL (Hashicorp Config Language) or JSON, a simplified version of this process could occur via Python dictionaries with JSON marshaling.

Additionally, adding new Terraform modules should be easier to plugin to this dynamic cluster generation process. It would be ideal if entire clusters could be described in variables.json verses one-off variables like s3_event_buckets.

Desired Outcome

• Express the entire StreamAlert infrastructure in variables.json in the form of Terraform module names.
• Generate all cluster files as JSON instead of HCL (Terraform can understand either).

Improvement: Optional Top Level Keys in a Schema and Parser Option Overhaul

Background

It is becoming increasingly common for certain data types to have inconsistent top level keys. This is also a familiar scenario during external API level changes where certain keys may or may not be present in data. An example of this would be:

type 1:
{"foo": 1, "bar": 2}

changing to

type 2:
{"foo": 1, "bar": 2, "baz": 3}

To support this behavior, a new option will be added to the log declaration in conf/logs.json.

Desired Outcome

Create a new setting in the log declaration called configuration, meant for various parser options:

setting	parser(s)	type	description
optional_top_level_keys	json	Map	Key/value pairs of optional fields with their type. If an incoming record does not have the key, a default value will be provided based on the declared type.
jsonpath	json	String	The JSONPath expression used to identify and parse nested records.
envelope_keys	json	Map	When using nested records, there may be top level keys we want to include in the record. This setting adds key/value pairs to a key called envelope in a record.
delimiter	csv, kv	String	The delimiter of fields in records.
separator	kv	String	The separator between keys and values in records.

Example

  "carbonblack:ingress.event.procstart": {
    "schema": {
      "cb_server": "string",
      "command_line": "string",
      "computer_name": "string",
      "event_type": "string",
      "expect_followon_w_md5": "boolean",
      "md5": "string",
      "parent_create_time": "integer",
      "parent_md5": "string",
      "parent_path": "string",
      "parent_process_guid": "string",
      "path": "string",
      "pid": "integer",
      "process_guid": "string",
      "sensor_id": "integer",
      "timestamp": "integer",
      "type": "string",
      "username": "string"
    },
    "parser": "json",
    "configuration": {
      "optional_top_level_keys": {
        "parent_md5": "string"
      }
    }
  }

Improvement: investigate PagerDuty incident merging

Updating this issue since the de-duping via the incident_key was not quite what we were looking for. As an alternative, we should investigate adding support for incident merging through the PD API:
https://v2.developer.pagerduty.com/v2/page/api-reference#!/Incidents/put_incidents_id_merge

Some implications of this that may not be ideal:

• A new incident would have to be created initially, which will probably result in another PD alert that we're trying to avoid.
• After the incident is created, we'd have to query incidents that exist with this incident_key and then merge any open incidents into one (presumably the oldest open incident):
  • https://v2.developer.pagerduty.com/v2/page/api-reference#!/Incidents/get_incidents

Other Notes:

• We could potentially combine something like the rule_name with the host_identifier or some other unique key to try to help with merging alerts that are from different hosts.
• PD is investigating adding a table like the new Alerts table found on pagerduty.com/alerts to the Details section of an incident.
• In the future, there could be a 'Page Once' feature from PD that must be opted-in on a per-incident basis that would aggregate all alerts after the mode is enabled.

Other Reference
https://support.pagerduty.com/hc/en-us/articles/230090307-Merging-Incidents

------Below is the original issue description for this issue------
PagerDuty supports automated de-duplication by including an incident_key with incidents. We should utilize this to reduce "alert fatigue" and "queue explosion" by setting the 'incident_key' to the StreamAlert rule name ('rule_name'). The incidents should then be merged within PagerDuty.

Reference
https://v2.developer.pagerduty.com/docs/events-api
https://support.pagerduty.com/hc/en-us/articles/202889630-How-do-I-reduce-the-number-of-notifications-I-am-receiving-for-the-same-incident-event

Feature: Rule Descriptions via Docstrings

Idea: Add descriptions/context to rules in function docstrings which are sent along with alerts:

@rule('invalid_subnet',
      logs=['osquery'],
      matchers=['logged_in_users'],
      outputs=['pagerduty'])
def invalid_subnet_rule(rec):
    """Catch logins from unauthorized subnets"""
    valid_cidr = IPNetwork('10.2.0.0/24')
    ip = IPAddress(rec['columns']['host'])

    return ip not in valid_cidr

Improved alert structure in rules_engine.py:

alert = {
  'rule_name': rule.rule_name,
  'rule_description': rule.rule_function.__doc__,
  'payload': payload,
  'outputs': rule.outputs
}

Feature: Support SNS and SQS as Sources

Background

AWS Lambda functions can be invoked from many different sources. Currently in StreamAlert, only AWS Kinesis and Amazon S3 are supported. In order to support a SOA based approach, allowing SNS and SQS to invoke StreamAlert's Lambda function will provide additional entry points to StreamAlert.

Desired Outcome

• Terraform module(s) which configure SNS/SQS permissions and event mappings
• Testing workflows

Improvement: Visit 'admin' privileges

Context:

• https://streamalert.readthedocs.io/account.html#administrator
• https://streamalert.slack.com/archives/general/p1486614692000033

Bug: Only require Terraform for infrastructure related CLI commands

Currently, terraform must be installed regardless of which CLI commands you run. Change this logic to only check if you run a relevant ./stream_alert_cli.py terraform ... command.

Improvement: Rule naming

Instead of defining the name of the rule function and passing the name as a string, pull the name of the function from the method.__name__ variable.

Bug: Identify KV with a Regular Expression

Currently the KV parser only supports a specific type of format:

key=value keyn=valuen...

The reason is that we split on the pair delimiter. The issue with this approach is that if your value contains that separator, the parser will fail. One way around this is to identify all pairs with a dynamic regex (based on the configured separators/delimiters), and then parse into a record.

Improvement: Add default values for separator/delimiter

For the KV and CSV parsers, add default values for the separators/delimiters if not specified from the config. Currently the parsing would error if they are omitted in the config.

Feature: AWS Step Functions

https://aws.amazon.com/step-functions/

Have the ability for alerts to go into step functions for decoration, context fetching or to make desired state change

Feature: Support email as an output

Support email as an output option (https://www.streamalert.io/outputs.html)

https://aws.amazon.com/ses/

Improvement: Ingest Cloudwatch Error Logs

Background

The tf_stream_alert_monitoring Terraform module creates several cloudwatch alarms to monitor AWS Lambda errors via defined thresholds. The issue with this approach is that there is no context to understand why a Lambda function is erring, and this must be triaged by hand.

Desired Outcome

Create a Cloudwatch Log subscription to Kinesis with a filter of Error|ERROR, and write StreamAlert rules to send alerts.

RFC: Logging in a different formal, e.g., Forensic Lucid

Hi,

Thanks for your Enigma talk and release of the tool.

I would like to change the logging format to Forensic Lucid in my fork in a modular way for investigation purposes and automated reasoning. What would be a better place to extend StreamAlert in that way?

Subsequently, if the code is sufficiently modular, how do I contribute the logger/formatter to you upstream? Do you have some contributor guidelines?

Thanks.

test

test

Feature: Threat intelligence (IOCs)

Target: Q3'17

StreamAlert will support comparing logs against traditional indicators of compromise (IOCs), which can range from thousands to millions in volume. This will be built in a way that's provider agnostic, allowing you to use ThreatStream, ThreatExchange, or whatever your heart desires.

Bug: Partition Large SNS Messages

Background

StreamAlert is divided into two AWS Lambda functions, one called alert and one called output. The bridge between both is AWS SNS. Records are first processed in the alert Lambda function, and if alerts are detected, it sends a JSON marshaled string to the SNS topic which invokes the output Lambda function.

Bug Info

Currently, if a large number of alerts is generated, the alert lambda function will throw a SNSMessageSizeError Exception because the marshaled string is too large:

SNS message size is too big! (Max: 256KB): SNSMessageSizeError
Traceback (most recent call last):
File "/var/task/main.py", line 91, in handler
StreamSink(alerts_to_send, config, env).sink()
File "/var/task/stream_alert/sink.py", line 60, in sink
self.publish_message(client, encodedSnsMessage, topic_arn)
File "/var/task/stream_alert/sink.py", line 117, in publish_message
raise SNSMessageSizeError('SNS message size is too big! (Max: 256KB)')
SNSMessageSizeError: SNS message size is too big! (Max: 256KB)

Desired Outcome

When publishing alerts to SNS, they should be partitioned into 256KB chunks and sent in multiple SNS messages. Offending code.

Improvement: Support List, Float, and Boolean Schema Types

Background

When declaring StreamAlert schemas for log types, a user can currently only specify the following values:

"key": "string",
"key2": "integer",
"key3": {}
"key4": {
  "nestedKey": "string"
}

The schema converts a parsed payload's values into their declared type as shown above. {} is a special type which accepts any Map values, which is representative of a 'loose schema'. Alternatively, strict schema checking will occur when nested keys are declared within a Map type.

Desired Outcome

Support boolean, float, and [] options for schema field types. [] should have the same behavior as {}.

Feature: Historical Search

Target: Q3/Q4'17

For historical searching, StreamAlert will use AWS Athena, a serverless, interactive query service that uses Presto to query data in S3. This will allow you to analyze data using SQL for both ad-hoc and scheduled queries.

Improvement: Integration Testing Helpers

Background

Some rules require time based fields, which evaluate to the last X hours (like last_hour).

Desired Outcome

When writing test logs, a new helper function which dynamically populates times in fields would solve this issue. For example:

{
  "records": [
    {
      "data": {
        "name": "logins",
        "host": "host1.test.net",
        "data": {
          "unixtime": "<helper:time_last_hour>",
          "user": "john"
        }
      },
      "description": "login from unauthorized users",
      "trigger": true,
      "source": "prefix_cluster1_stream_alert_kinesis",
      "service": "kinesis"
    }
  ]
}

Bug: StreamPayload Class SNSMessageSizeError

Currently, the StreamPayload class in the classifier.py contains methods and context around mapping incoming records to their source. It also contains attributes about parsed records such as raw_record, type, entity, and more.

As a result, when the class is Marshaled (in sink.py), too many instance attributes are attempting to be passed to the output Lambda function JSON blob. This class should be split into two: one for containing just the Payload, and one for processing incoming records.

[CLI] Pin Terraform Version

Pin the version of Terraform required by the StreamAlert CLI.

Improvement: Publish the python library separate to the Reference Implementation

So I tried to implement StreamAlert today. As someone with pretty good experience in AWS I felt like the "tools" used to make things easy hindered me rather than helped me. The python library is ideally all I need, I can build the rest and do it using the tools I prefer to use instead of trying to understand how the existing implementation is built.

Ideally the RuleHelper, RuleEngine and Classifier are all that are required for the simplest python library. The rest (including the Sink function) are more a Reference Implementation than a part of the core library, there are various ways to implement them depending on your requirements.

My strong but weakly held opinion is that this repository blurs the lines between the StreamAlert core rules engine and the infrastructure implementation; they are too tightly coupled. For example:

• The same configuration that feeds the Classifier is used to generate the Terraform manifest.
• The sink function is part of the same code base as the function that generates the alert.
• There's a function that packages and deploys Lambda functions, when many tools already exist that do this (Apex and Serverless)

Don't get me wrong, I 😍 that you've provided the automation example of how to implement this and I wish more OSS projects would do it too.

Bug: Documentation search is broken

Reproduction steps:

Load https://www.streamalert.io/search.html?q=account&check_keywords=yes&area=default
Click on any result

Details:

The links oddly end in .rst.html

Example: https://www.streamalert.io/account.rst.html

It should be https://www.streamalert.io/account.html

Fix:

Waiting on readthedocs/readthedocs.org#2708

Feature: HTTP Endpoint Support

StreamAlert will also support receiving data via an HTTP endpoint. This is for service providers or appliances that support HTTP endpoints for logging. Example: Akamai, OneLogin: https://support.onelogin.com/hc/en-us/articles/215214143-Streaming-Real-Time-OneLogin-Event-Data-to-your-SIEM-Solution

Python - third party libraries

We just missed shipping this for Usenix Enigma. Jack will update this issue with the plan and updates, it will ship soon

Feature: Crowdsource Plugin

Target: Q1/Q2'17

The idea of crowdsourcing your alerts isn't new. Slack does this and the blog speaks at length to the benefits. In the near future, StreamAlert will support this use-case, allowing you to decentralize your triage efforts, getting alerts to those with the most context. We’re aiming for Q1/Q2'17.

Improvement: testing should return an error value on any failure, but only after all tests run

When tests for StreamAlert fail, we current break and exit mid-testing. This make for a less than desirable user experience when multiple tests fail that need to be fixed. We should print each failure as it occurs and then exit with an error code (1) at the very end.

Improvement: Slack mrkdwn

It seems mrkdwn is supported for Slack message formatting. This would make the JSON dumped alert look much prettier.

Improvement: Support Terraform Remote Statefiles

Convert local tfstate to remote tfstate via S3 buckets

Feature: Verify/Support AWS Data Pipeline error logs

Background

AWS Data Pipelines can log errors to an S3 bucket:

http://docs.aws.amazon.com/datapipeline/latest/DeveloperGuide/dp-error-logs.html

Goal

• Identify the log format
• Verify if StreamAlert's existing S3 support does support this use-case
• If yes, update the documentation and close this issue
• If not, put up a PR to add support (schema / parser) if feasible

Bug: if pip.main() fails, the lambda package should not be uploaded

If the pip.main(pip_command) call within stream_alert_cli/package.py fails, it does so silently, and the lambda package will continue to be uploaded. We should check for failures and stop the process.

pip.main() returns a non-zero error code on failure, so this should be easy to evaluate and handle.

Log schema - Suricata (network security monitor)

What:

• https://suricata-ids.org/
• Add log schemas to conf/logs.json
• http://suricata.readthedocs.io/en/latest/output/index.html

Docs: Add development/testing documentation

Add a Development section to the docs which outlines:

• PRs
• Unit testing
• Integration testing

Feature: Support S3 Testing

Background

When testing the StreamAlert Lambda functions, it's necessary to create fake event records to send to the main handler for analysis. With AWS Kinesis and SNS this is straightforward, since the event contains the data to be analyzed/alerted on. However with s3 events, the data must be fetched out of band, which increases the complexity of testing.

Desired Outcome

The ability to test s3 based rules:

{
  "records": [
    {
      "data": "data,in,my,fake,object",
      "description": "Privileged wget commands to non Airbnb assets",
      "trigger": true,
      "source": "my.s3.bucket",
      "key": "my.fake.object",
      "service": "s3"
    }
  ]
}

Feature: @disable option for rules

What:

• An @disable option (or something equivalent) for rules

Why:

• Sometimes rules need to be temporarily disabled vs. deleted
• Need a way to have tests behave as expected when a rule is disabled

Feature - VPC flow log support

Feature - VPC flow log support

Likely via http://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/Subscriptions.html?tag=vig-20#LambdaFunctionExample

Issue: Not all regions support firehose

Hey, Oblivious this is an AWS issue. The required kinesis firehouse resource's, are available in three regions: EU (Ireland), US East (N. Virginia) and US West (Oregon)

Advice, tips or workarounds welcome!

Cheers,
Andrew

Improvement: Package dependencies for lambda

We already have code to package and deploy third party libraries, but it might be cool to have SA package its own dependencies (from requirements.txt) into the deployment archive as well. #40 introduced a new lambda dependency, and other revisions likely will too.

We could track these in third_party_libs in variables.json but IMO those should just be the dependencies of the rules themselves rather than their underlying infrastructure.

Log schema - Bro (network security monitor)

What:

• https://www.bro.org/
• Add log schemas to conf/logs.json
• https://www.bro.org/sphinx/script-reference/log-files.html

Improve: Make 'outputs' modular

Background

StreamAlert comes with an alerting framework: Rules can send alerts to one or many outputs.

Out of the box, StreamAlert supports:

• S3
• PagerDuty
• Slack

Goal:

While the code can be extended to support any API, 'outputs' needs to be more modular in the near to better support additional outputs and public contributions.

Feature: Custom CSV Delimiters

The ability to specify custom delimiters for CSV parsing, similar to how the KV parser works.

Desired outcome:

  "csv_log_name": {
    "schema": {
      "date": "string",
      "time": "integer",
      "host": "string",
      "message": "string"
    },
   "delimiter": "|",
    "hints": {
      "message": ["*keyword*"]
    }
  }

Bug: Matchers should act as a Rule constraint

Desired behavior: If a matcher evaluates to False, the code in the Rule does not execute. This is what's documented in our docs: https://www.streamalert.io/rules.html#matchers

Current behavior: Rule evaluates (A), matcher(s) evaluate (B), what's returned is A && B

Improvement: Rules Testing

Background: https://www.streamalert.io/rules.html#rules-testing

Goals

• Each rule has it's own .py for testing vs one big trigger_events.json and non_trigger_events.json
• Those .py's support whitespacing (ex: JSON log over multiple lines) vs. requiring one event per line

Bug: skip file if contents cannot be decoded into valid json

In stream_alert_cli/test.py, the function for testing kinesis alert rules (test_kinesis_alert_rules) should skip each file that cannot be properly decoded into json. Currently, the below error is thrown when a binary file (.DS_Store) is in the directory we're testing against.

Command
stream_alert_cli.py lambda test --func alert --source kinesis

Error

INFO:StreamAlertCLI:Error loading .DS_Store, bad JSON
Traceback (most recent call last):
  File "./stream_alert_cli.py", line 128, in <module>
    if __name__ == "__main__": main()
  File "./stream_alert_cli.py", line 125, in main
    cli.run(options)
  File "/Users/<user>/code/streamalert/stream_alert_cli/cli.py", line 59, in run
    self._lambda_runner(options)
  File "/Users/<user>/code/streamalert/stream_alert_cli/cli.py", line 73, in _lambda_runner
    stream_alert_test(options)
  File "/Users/<user>/code/streamalert/stream_alert_cli/test.py", line 216, in stream_alert_test
    test_kinesis_alert_rules()
  File "/Users/<user>/code/streamalert/stream_alert_cli/test.py", line 189, in test_kinesis_alert_rules
    test_records = contents['records']
UnboundLocalError: local variable 'contents' referenced before assignment

Feature: Outputs: JIRA

Support JIRA as an output

Bug: Alert Message Size

Background

Depending on the data source, alert size can vary. Before sending to Slack (or any input for that matter), message size should be taken into account and handled appropriately.

Desired Outcome

For Slack, the message size should be measured prior to sending. From the documentation, it mentions:

For best results, limit the number of characters in the text field to a few thousand bytes at most. Ideally, messages should be short and human-readable, if you need to post longer messages, please consider uploading a snippet instead. (A single message should be no larger than 4,000 bytes.)

Improvement: create a new CSV parser or update the current one to support input with a header row

The current CSVParser class is not designed to properly handle input if it contains a header row.
We should either subclass this parser or create a new one that can handle parsing of csv input that contains a header row. This parser should probably also try to validate the header to ensure the values match the keys we expect to receive. Python's csv.DictReader can likely be used for this, or we may be able to convert the CSVParser to strictly use csv.DictReader instead of adding a new class (will need to investigate more...)

Improvement: Combine Staging/Production Lambda Aliases

Background

Currently there are two AWS Lambda functions which make up StreamAlert:
alert: rules processing (rule matching and classification)
output: output processing (alerting to desired incident management/comms)

They have separate AWS Lambda Alias configurations:
alert has both staging and production
output only has staging

The difference between the two is that staging always points to the $LATEST uploaded StreamAlert source code S3 while production is a pointer to an immutable published version.

The difference from within StreamAlert is that staging does not send detected alerts to assigned outputs, it instead writes them to Cloudwatch logs. Staging historically has been used as a smoke test deployment to identify issues early on before it is ran against production data. However as this project has matured, and continuous integration testing has increased its coverage, we have less of a need for the split.

Desired Outcome

To simplify the deployment process, all StreamAlert Lambda functions should have a single alias: production.

This provides the following:

• A single, immutable version which can be rolled back as needed.
• No split application logic between staging/production.

Resistance to pentest? 33C3

Hello,

Thanks for sharing this promising tool.

How resistant is it against attacks exposed at 33C3 Gone in 60 seconds?
Gone in 60 Milliseconds, Intrusion and Exfiltration in Server-less Architectures
https://media.ccc.de/v/33c3-7865-gone_in_60_milliseconds

Things like shared temp in lambda, AWS IAM rules, exfiltration through meta
code still not released on https://github.com/Miserlou/Mackenzie/