Add Webex incoming webhook alerter

CHANGELOG.md

@@ -5,6 +5,7 @@
 
 ## New features
 - [MicrosoftPowerAutomate] Add support for 'ms_power_automate_webhook_url_from_field' option to dynamically select the webhook URL from the match. - [#1623](https://github.com/jertel/elastalert2/pull/1623) - @aizerin
+- Add Webex Incoming Webhook alerter - [#1635](https://github.com/jertel/elastalert2/pull/1635) - @dennis-trapp
 
 ## Other changes
 - Fix `schema.yaml` to support Kibana 8.17 - [#1631](https://github.com/jertel/elastalert2/pull/1631) - @vpiserchia

docs/source/alerts.rst

@@ -54,6 +54,7 @@ or
       - tencent_sms
       - twilio
       - victorops
+      - webex_webhook
       - workwechat
       - zabbix
 
@@ -2524,6 +2525,30 @@ Example with SMS usage::
     twilio_auth_token: "abcdefghijklmnopqrstuvwxyz012345"
     twilio_account_sid: "ABCDEFGHIJKLMNOPQRSTUVWXYZ01234567"
 
+Webex Webhook
+~~~~~~~~~~~~~
+
+Webex Webhook alerter will send notification to a predefined incoming webhook in Webex application. The body of the notification is formatted the same as with other alerters.
+
+Official Webex incoming webhook documentation: https://apphub.webex.com/applications/incoming-webhooks-cisco-systems-38054-23307-75252
+
+Required:
+
+``webex_webhook_id``:  Webex webhook ID.
+``webex_webhook_msgtype``:  Webex webhook message format. Can be ``text`` or ``markdown``. Defaults to ``text``.
+
+Example usage::
+
+    alert_text: "**{0}** - ALERT on host {1}"
+    alert_text_args:
+      - name
+      - hostname
+    alert:
+      - webex_webhook
+    alert_text_type: alert_text_only
+    webex_webhook_id: "your webex incoming webhook id"
+    webex_webhook: "markdown"
+
 WorkWechat
 ~~~~~~~~~~
 

docs/source/elastalert.rst

@@ -66,6 +66,7 @@ Currently, we have support built in for these alert types:
 - Tencent SMS
 - TheHive
 - Twilio
+- Webex Incoming Webhook
 - WorkWechat  
 - Zabbix
 

elastalert/alerters/webex_webhook.py

@@ -0,0 +1,54 @@
+import json
+import warnings
+
+import requests
+from elastalert.alerts import Alerter, DateTimeEncoder
+from elastalert.util import EAException, elastalert_logger
+from requests import RequestException
+
+
+class WebexWebhookAlerter(Alerter):
+    """Creates a Webex Incoming Webook message for each alert"""
+
+    required_options = frozenset(["webex_webhook_id"])
+
+    def __init__(self, rule):
+        super(WebexWebhookAlerter, self).__init__(rule)
+        self.webex_webhook_id = self.rule.get(
+            "webex_webhook_id", None
+        )
+        self.webex_webhook_url = f"https://webexapis.com/v1/webhooks/incoming/{self.webex_webhook_id}"
+        self.webex_webhook_msgtype = self.rule.get("webex_webhook_msgtype", "text")
+
+    def alert(self, matches):
+        body = self.create_alert_body(matches)
+
+        headers = {
+            "Content-Type": "application/json",
+            "Accept": "application/json;charset=utf-8",
+        }
+
+        payload = {"text": body}
+
+        if self.webex_webhook_msgtype == "markdown":
+            payload = {"markdown": body}
+
+        try:
+            response = requests.post(
+                self.webex_webhook_url,
+                data=json.dumps(payload, cls=DateTimeEncoder),
+                headers=headers,
+            )
+            warnings.resetwarnings()
+            response.raise_for_status()
+        except RequestException as e:
+            raise EAException("Error posting to webex_webhook: %s" % e)
+
+        elastalert_logger.info("Trigger sent to webex_webhook")
+
+    def get_info(self):
+        return {
+            "type": "webex_webhook",
+            "webex_webhook_msgtype": self.webex_webhook_msgtype,
+            "webex_webhook_url": self.webex_webhook_url,
+        }

elastalert/loaders.py

@@ -37,6 +37,7 @@
 import elastalert.alerters.thehive
 import elastalert.alerters.twilio
 import elastalert.alerters.victorops
+import elastalert.alerters.webex_webhook
 import elastalert.alerters.workwechat
 from elastalert import alerts
 from elastalert import enhancements
@@ -134,6 +135,7 @@ class RulesLoader(object):
         'discord': elastalert.alerters.discord.DiscordAlerter,
         'dingtalk': elastalert.alerters.dingtalk.DingTalkAlerter,
         'lark': elastalert.alerters.lark.LarkAlerter,
+        'webex_webhook': elastalert.alerters.webex_webhook.WebexWebhookAlerter,
         'workwechat': elastalert.alerters.workwechat.WorkWechatAlerter,
         'chatwork': elastalert.alerters.chatwork.ChatworkAlerter,
         'datadog': elastalert.alerters.datadog.DatadogAlerter,

elastalert/schema.yaml

@@ -899,6 +899,10 @@ properties:
   twilio_message_service_sid: {type: string}
   twilio_use_copilot: {type: boolean}
 
+  ### Webex Webhook
+  webex_webhook_id: { type: string }
+  webex_webhook_msgtype: { type: string }
+
   ### WorkWechat
   work_wechat_bot_id: { type: string }
   work_wechat_msgtype: { type: string }

tests/alerters/webex_webhook_test.py

@@ -0,0 +1,123 @@
+import json
+import logging
+from unittest import mock
+
+import pytest
+from requests import RequestException
+
+from elastalert.alerters.webex_webhook import WebexWebhookAlerter
+from elastalert.loaders import FileRulesLoader
+from elastalert.util import EAException
+
+
+def test_webex_webhook_text(caplog):
+    caplog.set_level(logging.INFO)
+    rule = {
+        'name': 'Test Webex Webhook Rule',
+        'type': 'any',
+        'webex_webhook_msgtype': 'text',
+        'webex_webhook_id': 'xxxxxxx',
+        'alert': [],
+    }
+    rules_loader = FileRulesLoader({})
+    rules_loader.load_modules(rule)
+    alert = WebexWebhookAlerter(rule)
+    match = {
+        '@timestamp': '2024-01-30T00:00:00',
+        'somefield': 'foobar'
+    }
+    with mock.patch('requests.post') as mock_post_request:
+        alert.alert([match])
+
+    expected_data = {
+        'text': 'Test Webex Webhook Rule\n\n@timestamp: 2024-01-30T00:00:00\nsomefield: foobar\n'
+    }
+
+    mock_post_request.assert_called_once_with(
+        'https://webexapis.com/v1/webhooks/incoming/xxxxxxx',
+        data=mock.ANY,
+        headers={
+            'Content-Type': 'application/json',
+            'Accept': 'application/json;charset=utf-8'
+        }
+    )
+
+    actual_data = json.loads(mock_post_request.call_args_list[0][1]['data'])
+    assert expected_data == actual_data
+    assert ('elastalert', logging.INFO, 'Trigger sent to webex_webhook') == caplog.record_tuples[0]
+
+
+def test_webex_webhook_ea_exception():
+    with pytest.raises(EAException) as ea:
+        rule = {
+            'name': 'Test Webex Webhook Rule',
+            'type': 'any',
+            'webex_webhook_msgtype': 'text',
+            'webex_webhook_id': 'xxxxxxx',
+            'alert': [],
+        }
+        rules_loader = FileRulesLoader({})
+        rules_loader.load_modules(rule)
+        alert = WebexWebhookAlerter(rule)
+        match = {
+            '@timestamp': '2024-01-30T00:00:00',
+            'somefield': 'foobar'
+        }
+        mock_run = mock.MagicMock(side_effect=RequestException)
+        with mock.patch('requests.post', mock_run), pytest.raises(RequestException):
+            alert.alert([match])
+    assert 'Error posting to webex_webhook: ' in str(ea)
+
+
+def test_webex_webhook_getinfo():
+    rule = {
+        'name': 'Test Webex Webhook Rule',
+        'type': 'any',
+        'webex_webhook_msgtype': 'text',
+        'webex_webhook_id': 'xxxxxxx',
+        'alert': [],
+    }
+    rules_loader = FileRulesLoader({})
+    rules_loader.load_modules(rule)
+    alert = WebexWebhookAlerter(rule)
+
+    expected_data = {
+        'type': 'webex_webhook',
+        'webex_webhook_msgtype': 'text',
+        'webex_webhook_url': 'https://webexapis.com/v1/webhooks/incoming/xxxxxxx'
+    }
+    actual_data = alert.get_info()
+    assert expected_data == actual_data
+
+
+@pytest.mark.parametrize('webex_webhook_id, webex_webhook_msgtype, expected_data', [
+    ('', '', 'Missing required option(s): webex_webhook_id, webex_webhook_msgtype'),
+    ('xxxxxxx', 'yyyyyy',
+     {
+        'type': 'webex_webhook',
+        'webex_webhook_msgtype': 'yyyyyy',
+        'webex_webhook_url': 'https://webexapis.com/v1/webhooks/incoming/xxxxxxx'
+     }),
+])
+def test_webex_webhook_required_error(webex_webhook_id, webex_webhook_msgtype, expected_data):
+    try:
+        rule = {
+            'name': 'Test Webex Webhook Rule',
+            'type': 'any',
+            'alert': [],
+        }
+
+        if webex_webhook_id:
+            rule['webex_webhook_id'] = webex_webhook_id
+
+        if webex_webhook_msgtype:
+            rule['webex_webhook_msgtype'] = webex_webhook_msgtype
+
+        rules_loader = FileRulesLoader({})
+        rules_loader.load_modules(rule)
+        alert = WebexWebhookAlerter(rule)
+
+        actual_data = alert.get_info()
+        assert expected_data == actual_data
+    except Exception as ea:
+        assert expected_data in str(ea)