Skip to content

Elastalert Custom Modules

Abstract

Targetted audience

Standard elastalert rules are not enough for satisfying your use-case.

Coding your custom Rules

Extending Elastalert Rules API, Alert API, Enhancement API and Rules Loader API enables you to write more specific business logic to answer your problems.

PEX Project setup

We provide a starter on our github repository

setup.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-


from setuptools import setup, find_packages

setup(
    name="punch_plugins",
    version="1.0",
    packages=find_packages(),
    include_package_data=True,
    author="punchplatform",
    author_email="contact@punchplatform.com",
    description="boilerplate for elastaler custom alerter, rule type, etc...",
    python_requires='>=3.6',
    install_requires=[
        "pex==2.1.6",  # this is important do not remove
        "requests==2.24.0",  # this is important do not remove
        "pymongo",
    ]
)

python package hierarchy

├── example_configs
│   ├── alert
│   │   ├── config.yaml
│   │   └── rule.yaml
│   ├── enhancement
│   │   ├── config.yaml
│   │   └── rule.yaml
│   ├── rule
│   │   ├── config.yaml
│   │   └── rule.yaml
│   └── rule_loader
│       ├── config.yaml
│       └── rule.yaml
├── Makefile
├── punch_plugins
│   ├── alerters
│   │   ├── __init__.py
│   │   └── my_use_case
│   │       ├── complex_alert.py
│   │       └── __init__.py
│   ├── enhancements
│   │   ├── __init__.py
│   │   └── my_use_case
│   │       ├── complex_enhancement.py
│   │       └── __init__.py
│   ├── __init__.py
│   ├── rules
│   │   ├── __init__.py
│   │   └── my_use_case
│   │       ├── complex_rule.py
│   │       └── __init__.py
│   └── rules_loader
│       ├── __init__.py
│       └── my_use_case
│           ├── complex_rule_loader.py
│           └── __init__.py
├── README.md
└── setup.py

Rule Module

A simple custom rule module taken for Elastalert official documentation:

#!/usr/bin/env python3
# -*- coding: utf-8 -*-


import dateutil.parser

from elastalert.ruletypes import RuleType

# elastalert.util includes useful utility functions
# such as converting from timestamp to datetime obj
from elastalert.util import ts_to_dt  # NOQA


class ComplexRule(RuleType):

    # By setting required_options to a set of strings
    # You can ensure that the rule config file specifies all
    # of the options. Otherwise, ElastAlert will throw an exception
    # when trying to load the rule.
    required_options = set(["time_start", "time_end", "usernames"])

    # add_data will be called each time Elasticsearch is queried.
    # data is a list of documents from Elasticsearch, sorted by timestamp,
    # including all the fields that the config specifies with "include"
    def add_data(self, data):
        for document in data:

            # To access config options, use self.rules
            if document["username"] in self.rules["usernames"]:

                # Convert the timestamp to a time object
                login_time = document["@timestamp"].time()

                # Convert time_start and time_end to time objects
                time_start = dateutil.parser.parse(self.rules["time_start"]).time()
                time_end = dateutil.parser.parse(self.rules["time_end"]).time()

                # If the time falls between start and end
                if login_time > time_start and login_time < time_end:

                    # To add a match, use self.add_match
                    self.add_match(document)

    # The results of get_match_str will appear in the alert text
    def get_match_str(self, match):
        return "%s logged in between %s and %s" % (
            match["username"],
            self.rules["time_start"],
            self.rules["time_end"],
        )

    # garbage_collect is called indicating that ElastAlert has already been run up to timestamp
    # It is useful for knowing that there were no query results from Elasticsearch because
    # add_data will not be called with an empty list
    def garbage_collect(self, timestamp):
        pass

Alert Module

A simple custom alert module taken for Elastalert official documentation:

#!/usr/bin/env python3
# -*- coding: utf-8 -*-


from elastalert.alerts import Alerter, BasicMatchString

class ComplexAlert(Alerter):

    # By setting required_options to a set of strings
    # You can ensure that the rule config file specifies all
    # of the options. Otherwise, ElastAlert will throw an exception
    # when trying to load the rule.
    required_options = set(['output_file_path'])

    # Alert is called
    def alert(self, matches):

        # Matches is a list of match dictionaries.
        # It contains more than one match when the alert has
        # the aggregation option set
        for match in matches:

            # Config options can be accessed with self.rule
            with open(self.rule['output_file_path'], "a") as output_file:

                # basic_match_string will transform the match into the default
                # human readable string format
                match_string = str(BasicMatchString(self.rule, match))

                output_file.write(match_string)

    # get_info is called after an alert is sent to get data that is written back
    # to Elasticsearch in the field "alert_info"
    # It should return a dict of information relevant to what the alert does
    def get_info(self):
        return {'type': 'Awesome Alerter',
                'output_file': self.rule['output_file_path']}

Enhancement Module

A simple custom enhancement module taken for Elastalert official documentation:

#!/usr/bin/env python3
# -*- coding: utf-8 -*-


from elastalert.enhancements import BaseEnhancement

class ComplexEnhancement(BaseEnhancement):

    # The enhancement is run against every match
    # The match is passed to the process function where it can be modified in any way
    # ElastAlert will do this for each enhancement linked to a rule
    def process(self, match):
        if 'domain' in match:
            url = "http://who.is/whois/%s" % (match['domain'])
            match['domain_whois_link'] = url

Rule Loader Module

A simple custom rule loader module taken for Elastalert official documentation:

#!/usr/bin/env python3
# -*- coding: utf-8 -*-


from pymongo import MongoClient
from elastalert.loaders import RulesLoader
import yaml

class ComplexRuleLoader(RulesLoader):
    def __init__(self, conf):
        super(ComplexRuleLoader, self).__init__(conf)
        self.client = MongoClient(conf['mongo_url'])
        self.db = self.client[conf['mongo_db']]
        self.cache = {}

    def get_names(self, conf, use_rule=None):
        if use_rule:
            return [use_rule]

        rules = []
        self.cache = {}
        for rule in self.db.rules.find():
            self.cache[rule['name']] = yaml.load(rule['yaml'])
            rules.append(rule['name'])

        return rules

    def get_hashes(self, conf, use_rule=None):
        if use_rule:
            return [use_rule]

        hashes = {}
        self.cache = {}
        for rule in self.db.rules.find():
            self.cache[rule['name']] = rule['yaml']
            hashes[rule['name']] = rule['hash']

        return hashes

    def get_yaml(self, rule):
        if rule in self.cache:
            return self.cache[rule]

        self.cache[rule] = yaml.load(self.db.rules.find_one({'name': rule})['yaml'])
        return self.cache[rule]

Deploying your custom module

Installing your module

Standalone mode

For the standalone, put your pexs in $PUNCHPLATFORM_INSTALL_DIR/extlib/elastalert/.

Deployed mode

For a deployed mode, refer to this documentation

Using your custom modules

config.yaml

es_host: localhost
es_port: 9200
writeback_index: elastalert
run_every:
  minutes: 2
buffer_time:
  minutes: 1
rules_folder: examples

rule.yaml

es_host: localhost
es_port: 9200
name: "Punch Example Rule"
type: "punch_plugins.rules.my_use_case.complex_rule.ComplexRule"
index: logstash-*
num_events: 50
timeframe:
    hours: 4
filter:
- term:
    some_field: "some_value"
alert:
- "email"
email:
- "contact@punchplatform.com"
time_start: "20:00"
usernames:
- "admin"
- "userXYZ"
- "foobaz"
time_end: "24:00"

Execution

WORK_DIR=$(pwd)

ls $WORK_DIR
example_configs/rule/config.yaml
example_configs/rule/rule.yaml

punchplatform-elastalert.sh \
                --start-foreground \
                --config $WORK_DIR/example_configs/rule/config.yaml \
                --rule example_configs/rule/rule.yaml \
                --additional-pex custom_rule-1.0.0.pex