Added new alerter to send alerts to Opensearch

CHANGELOG.md

@@ -4,7 +4,7 @@
 - TBD
 
 ## New features
-- TBD
+- Add indexer alerter - [#1451](https://github.com/jertel/elastalert2/pull/1451) - @olehpalanskyi
 
 ## Other changes
 - [Docs] Fixed typo in Alerta docs with incorrect number of seconds in a day. - @jertel

docs/source/alerts.rst

@@ -28,6 +28,7 @@ or
       - googlechat
       - gelf
       - hivealerter
+      - indexer
       - iris
       - jira
       - lark
@@ -49,7 +50,7 @@ or
       - tencent_sms
       - twilio
       - victorops
-      - workwechat  
+      - workwechat
       - zabbix
 
 Options for each alerter can either defined at the top level of the YAML file, or nested within the alert name, allowing for different settings
@@ -1085,8 +1086,91 @@ Example usage with json string formatting::
         "X-custom-{{key}}": "{{type}}"
       }
 
+Indexer
+~~~~~~~
+
+Description: Creates a record in an arbitrary index within an Elasticsearch or OpenSearch index.
+
+Indexer alerter can be used to create a new alert in existing Opensearch/Elasticsearch. The alerter supports
+custom fields, and observables from the alert matches and rule data.
+
+Required:
+
+``indexer_alert_config``: Configuration options for the alert, see example below for structure.
+
+``customFields`` Fields must be manually added, all of them will exist in the newly created index. You can set own field or use existing field fron match (see example below for structure).
+
+``indexer_alerts_name``: The index to use for creating the new alert records.
+
+One of below is required:
+
+``indexer_connection``: Options the connection details to your server instance (see example below for the required syntax Example 1).
+
+``indexer_config``: Options for loading the connection details to your server instance from a file (see example below for the required syntax Example 2).
+
+
+Example 1 usage::
+
+    alert: indexer
+
+    indexer_connection:
+      es_host: localhost
+      es_port: es_port
+      ssl_show_warn: False
+      use_ssl: True
+      verify_certs: False
+      es_username: user
+      es_password: password
+      indexer_alerts_name: elastalert2               # You can create own config or use global config just added ``indexer_alerts_name`` in global config
+
+    indexer_alert_config:
+      #Existing fields from match alert
+      message: message
+      host.name: host.name
+      event.action: event.action
+      event.type: event.type
+      winlog.computer_name: winlog.computer_name
+      winlog.event_id: winlog.event_id
+      winlog.task: winlog.task
+      #Enrich existing event with additional fields
+      customFields:
+        - name: original_time
+          value: "@timestamp"
+        - name: severity
+          value: high
+        - name: risk_score
+          value: 73
+        - name: description
+          value: General description.
+
+Example 2 usage::
+
+    alert: indexer
+
+    indexer_config: /opt/elastalert/config/config.yaml       # Uses the ElastAlert 2 global config, with an added ``indexer_alerts_name`` parameter
+
+    indexer_alert_config:
+      #Existing fields from match alert
+      message: message
+      host.name: host.name
+      event.action: event.action
+      event.type: event.type
+      winlog.computer_name: winlog.computer_name
+      winlog.event_id: winlog.event_id
+      winlog.task: winlog.task
+      #Enrich existing event with additional fields
+      customFields:
+        - name: original_time
+          value: "@timestamp"
+        - name: severity
+          value: high
+        - name: risk_score
+          value: 73
+        - name: description
+          value: General description.
+
 IRIS
-~~~~~~~~~
+~~~~
 The Iris alerter can be used to create a new alert or case in `Iris IRP System <https://dfir-iris.org>`_. The alerter supports adding tags, IOCs, and context from the alert matches and rule data.
 
 The alerter requires the following option:

docs/source/elastalert.rst

@@ -44,6 +44,7 @@ Currently, we have support built in for these alert types:
 - Graylog GELF
 - HTTP POST
 - HTTP POST 2
+- Indexer
 - Iris
 - Jira
 - Lark

elastalert/alerters/indexer.py

@@ -0,0 +1,113 @@
+import os
+import yaml
+from datetime import datetime
+from elasticsearch.exceptions import TransportError
+from elastalert.alerts import Alerter
+from elastalert.util import lookup_es_key, EAException, elastalert_logger, elasticsearch_client
+
+class IndexerAlerter(Alerter):
+    """
+    Use matched data to create alerts on Opensearch/Elasticsearch
+    """
+    required_options = frozenset(['indexer_alert_config'])
+
+    def lookup_field(self, match: dict, field_name: str, default):
+        field_value = lookup_es_key(match, field_name)
+        if field_value is None:
+            field_value = self.rule.get(field_name, default)
+
+        return field_value
+
+    def get_query(self,body_request_raw):
+        original = body_request_raw[0]
+        for orig in original.values():
+            for query_string in orig.values():
+                query = query_string
+        return query['query']
+
+    def lookup_list_fields(self, original_fields_raw: list, match: dict):
+        original_fields = {}
+        for field in original_fields_raw:
+            if field.get('value'):
+                if (isinstance(field['value'], str)):
+                    if field['value'] == 'filter':
+                        body_request_raw = self.rule.get(field['value'])
+                        value = self.get_query(body_request_raw)
+                    else:
+                        value = self.lookup_field(match, field['value'], field['value'])
+                else:
+                    value = field['value']
+                original_fields[field['name']] = value
+            else:
+                for k,v in field.items():
+                    original_fields[k] = self.lookup_list_fields(v, match)
+
+        return original_fields
+
+    def event_orig_fields(self, original_fields_raw, match: dict):
+        if (isinstance(original_fields_raw, str)):
+            value = self.lookup_field(match, original_fields_raw, original_fields_raw)
+        elif (isinstance(original_fields_raw, list)):
+            value = self.lookup_list_fields(original_fields_raw, match)
+        else:
+            value = original_fields_raw
+        return value
+
+    def make_nested_fields(self, data):
+        nested_data = {}
+        for key, value in data.items():
+            keys = key.split(".")
+            current_nested_data = nested_data
+            for nested_key in keys[:-1]:
+                current_nested_data = current_nested_data.setdefault(nested_key, {})
+            current_nested_data[keys[-1]] = value
+        return nested_data
+
+    def flatten_dict(self, data, prefix='', sep='.'):
+        nd = {}
+        for k, v in data.items():
+            if isinstance(v, dict):
+                nd.update(self.flatten_dict(v, f'{prefix}{k}{sep}'))
+            else:
+                nd[f'{prefix}{k}'] = v
+        return nd
+
+    def remove_matching_pairs(self, input_dict):
+        return {key: value for key, value in input_dict.items() if key != value}
+
+    def alert(self, matches):
+        alert_config = {
+            '@timestamp': datetime.now().strftime('%Y-%m-%dT%H:%M:%S.%fZ')
+        }
+        alert_config.update(self.rule.get('indexer_alert_config', {}))
+
+        if len(matches) > 0:
+            alert_config = self.flatten_dict(alert_config)
+            for event_orig in alert_config:
+                alert_config[event_orig] = self.event_orig_fields(alert_config[event_orig],matches[0])
+        alert_config = self.remove_matching_pairs(self.flatten_dict(alert_config))
+        alert_config = self.make_nested_fields(alert_config)
+
+
+        # POST the alert to SIEM
+        try:
+            data = self.rule.get('indexer_connection', '')
+            if not data:
+                if os.path.isfile(self.rule.get('indexer_config', '')):
+                    filename = self.rule.get('indexer_config', '')
+                else:
+                    filename = ''
+
+                if filename:
+                    with open(filename) as config_file:
+                        data = yaml.load(config_file, Loader=yaml.FullLoader)
+            elasticsearch_client(data).index(index = data.get('indexer_alerts_name'),
+                                              body = alert_config,
+                                              refresh = True)
+
+        except TransportError as e:
+            raise EAException(f"Error posting to SIEM: {e}")
+        elastalert_logger.info("Alert sent to SIEM")
+
+    def get_info(self):
+        return {'type': 'indexer'}

elastalert/loaders.py

@@ -52,6 +52,7 @@
 from elastalert.alerters.teams import MsTeamsAlerter
 from elastalert.alerters.zabbix import ZabbixAlerter
 from elastalert.alerters.tencentsms import TencentSMSAlerter
+from elastalert.alerters.indexer import IndexerAlerter
 from elastalert.util import dt_to_ts
 from elastalert.util import dt_to_ts_with_format
 from elastalert.util import dt_to_unix
@@ -137,6 +138,7 @@ class RulesLoader(object):
         'rocketchat': elastalert.alerters.rocketchat.RocketChatAlerter,
         'gelf': elastalert.alerters.gelf.GelfAlerter,
         'iris': elastalert.alerters.iris.IrisAlerter,
+        'indexer': IndexerAlerter,
     }
 
     # A partial ordering of alert types. Relative order will be preserved in the resulting alerts list

elastalert/schema.yaml

@@ -538,6 +538,35 @@ properties:
   http_post2_ignore_ssl_errors: {type: boolean}
   http_post2_timeout: {type: integer}
 
+  ### INDEXER
+  indexer_alerts_name: {type: string}
+  indexer_config: {type: string}
+  indexer_connection:
+    type: object
+    properties:
+      es_host: {type: string}
+      es_hosts: {type: array, items: {type: string}}
+      es_port: {type: integer}
+      ssl_show_warn: {type: boolean}
+      use_ssl: {type: boolean}
+      verify_certs: {type: boolean}
+      ca_cert: {type: string}
+      es_username: {type: string}
+      es_password: {type: string}
+      index_alerts_name: {type: string}
+  indexer_alert_config:
+    type: object
+    minProperties: 1
+    patternProperties:
+      "^.+$":
+        oneOf:
+          - type: [boolean, string, integer]
+          - type: object
+            additionalProperties: false
+            required: [ field ]
+            properties:
+              field: { type: [boolean, string, integer], minLength: 1 }
+
   ### IRIS
   iris_host: {type: string}
   iris_api_token: {type: string}