add workwechat alerter

CHANGELOG.md

@@ -4,7 +4,7 @@
 - TBD
 
 ## New features
-- TBD
+- Add workwechat alerter - [#1367](https://github.com/jertel/elastalert2/pull/1367) - @wufeiqun
 
 ## Other changes
 - TBD

docs/source/alerts.rst

@@ -49,6 +49,7 @@ or
       - tencent_sms
       - twilio
       - victorops
+      - 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
@@ -2272,6 +2273,21 @@ Example with SMS usage::
     twilio_auth_token: "abcdefghijklmnopqrstuvwxyz012345"
     twilio_account_sid: "ABCDEFGHIJKLMNOPQRSTUVWXYZ01234567"
 
+WorkWechat
+~~~~~~~~~~
+
+WorkWechat alerter will send notification to a predefined bot in WorkWechat application. The body of the notification is formatted the same as with other alerters.
+
+Required:
+
+``work_wechat_bot_id``:  WorkWechat bot id.
+
+Example usage::
+
+    alert:
+      - "workwechat"
+    work_wechat_bot_id: "your workwechat bot id"
+
 Zabbix
 ~~~~~~
 

docs/source/elastalert.rst

@@ -63,6 +63,7 @@ Currently, we have support built in for these alert types:
 - Tencent SMS
 - TheHive
 - Twilio
+- WorkWechat  
 - Zabbix
 
 Additional rule types and alerts can be easily imported or written. (See :ref:`Writing rule types <writingrules>` and :ref:`Writing alerts <writingalerts>`)

elastalert/alerters/workwechat.py

@@ -0,0 +1,52 @@
+import json
+import warnings
+
+import requests
+from elastalert.alerts import Alerter, DateTimeEncoder
+from elastalert.util import EAException, elastalert_logger
+from requests import RequestException
+
+
+class WorkWechatAlerter(Alerter):
+    """ Creates a WorkWechat message for each alert """
+    required_options = frozenset(['work_wechat_bot_id'])
+
+    def __init__(self, rule):
+        super(WorkWechatAlerter, self).__init__(rule)
+        self.work_wechat_bot_id = self.rule.get('work_wechat_bot_id', None)
+        self.work_wechat_webhook_url = f'https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key={self.work_wechat_bot_id}'
+        self.work_wechat_msg_type = 'text'
+
+    def alert(self, matches):
+        title = self.create_title(matches)
+        body = self.create_alert_body(matches)
+
+        headers = {
+            'Content-Type': 'application/json',
+            'Accept': 'application/json;charset=utf-8'
+        }
+
+        payload = {
+            'msgtype': self.work_wechat_msg_type,
+            "text": {
+                "content": body
+            },
+        }
+
+        try:
+            response = requests.post(
+                self.work_wechat_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 workwechat: %s" % e)
+
+        elastalert_logger.info("Trigger sent to workwechat")
+
+    def get_info(self):
+        return {
+            "type": "workwechat",
+            "work_wechat_webhook_url": self.work_wechat_webhook_url
+        }

elastalert/loaders.py

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

elastalert/schema.yaml

@@ -793,6 +793,9 @@ properties:
   twilio_message_service_sid: {type: string}
   twilio_use_copilot: {type: boolean}
 
+  ### WorkWechat
+  work_wechat_bot_id: { type: string }
+
   ### Zabbix
   zbx_sender_host: {type: string}
   zbx_sender_port: {type: integer}

tests/alerters/workwechat_test.py

@@ -0,0 +1,118 @@
+import json
+import logging
+from unittest import mock
+
+import pytest
+from requests import RequestException
+
+from elastalert.alerters.workwechat import WorkWechatAlerter
+from elastalert.loaders import FileRulesLoader
+from elastalert.util import EAException
+
+
+def test_work_wechat_text(caplog):
+    caplog.set_level(logging.INFO)
+    rule = {
+        'name': 'Test WorkWechat Rule',
+        'type': 'any',
+        'work_wechat_bot_id': 'xxxxxxx',
+        'alert': [],
+    }
+    rules_loader = FileRulesLoader({})
+    rules_loader.load_modules(rule)
+    alert = WorkWechatAlerter(rule)
+    match = {
+        '@timestamp': '2024-01-30T00:00:00',
+        'somefield': 'foobar'
+    }
+    with mock.patch('requests.post') as mock_post_request:
+        alert.alert([match])
+
+    expected_data = {
+        'msgtype': 'text',
+        'text': {
+            'content': 'Test WorkWechat Rule\n\n@timestamp: 2024-01-30T00:00:00\nsomefield: foobar\n'
+        }
+    }
+
+    mock_post_request.assert_called_once_with(
+        'https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=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 workwechat') == caplog.record_tuples[0]
+
+
+def test_work_wechat_ea_exception():
+    with pytest.raises(EAException) as ea:
+        rule = {
+            'name': 'Test WorkWechat Rule',
+            'type': 'any',
+            'work_wechat_bot_id': 'xxxxxxx',
+            'alert': [],
+        }
+        rules_loader = FileRulesLoader({})
+        rules_loader.load_modules(rule)
+        alert = WorkWechatAlerter(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 workwechat: ' in str(ea)
+
+
+def test_work_wechat_getinfo():
+    rule = {
+        'name': 'Test WorkWechat Rule',
+        'type': 'any',
+        'work_wechat_bot_id': 'xxxxxxx',
+        'alert': [],
+    }
+    rules_loader = FileRulesLoader({})
+    rules_loader.load_modules(rule)
+    alert = WorkWechatAlerter(rule)
+
+    expected_data = {
+        'type': 'workwechat',
+        'work_wechat_webhook_url': 'https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxxxxxx'
+    }
+    actual_data = alert.get_info()
+    assert expected_data == actual_data
+
+
+@pytest.mark.parametrize('work_wechat_bot_id, expected_data', [
+    ('', 'Missing required option(s): work_wechat_bot_id'),
+    ('xxxxxxx',
+     {
+         'type': 'workwechat',
+         'work_wechat_webhook_url': 'https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxxxxxx'
+     }),
+])
+def test_work_wechat_required_error(work_wechat_bot_id, expected_data):
+    try:
+        rule = {
+            'name': 'Test WorkWechat Rule',
+            'type': 'any',
+            'alert': [],
+        }
+
+        if work_wechat_bot_id:
+            rule['work_wechat_bot_id'] = work_wechat_bot_id
+
+        rules_loader = FileRulesLoader({})
+        rules_loader.load_modules(rule)
+        alert = WorkWechatAlerter(rule)
+
+        actual_data = alert.get_info()
+        assert expected_data == actual_data
+    except Exception as ea:
+        assert expected_data in str(ea)