Merge pull request #1485 from innerpeacez/support-dingtalk-sign-send

Support DingTalk robot, add sign security settings

CHANGELOG.md

@@ -23,6 +23,7 @@
 - [workwechat] add workwechat msgtype - [#1369](https://github.com/jertel/elastalert2/pull/1369) - @bitqiu
 - [Pager Duty] Add options: pagerduty_ca_certs, pagerduty_ignore_ssl_errors - [#1418](https://github.com/jertel/elastalert2/pull/1418) - @kexin-zhai
 - Add support for Kibana 8.13 for Kibana Discover - [#1423](https://github.com/jertel/elastalert2/pull/1423) - @nsano-rururu
+- Support DingTalk robot, add sign security settings - [#1485](https://github.com/jertel/elastalert2/pull/1485) - @innerpeacez
 
 # 2.17.0
 

docs/source/alerts.rst

@@ -698,9 +698,11 @@ Optional:
 
 ``dingtalk_proxy``: By default ElastAlert 2 will not use a network proxy to send notifications to Dingtalk. Set this option using ``hostname:port`` if you need to use a proxy. only supports https.
 
-``dingtalk_proxy_login``: The Dingtalk proxy auth username.
+``dingtalk_proxy_login``: The DingTalk proxy auth username.
 
-``dingtalk_proxy_pass``: The Dingtalk proxy auth username.
+``dingtalk_proxy_pass``: The DingTalk proxy auth username.
+
+``dingtalk_sign``: DingTalk HMAC secret, used for message authentication. See https://open.dingtalk.com/document/robots/customize-robot-security-settings for more information. Note that the algorithm provides authentication that *some* message was recently sent (within an hour) but does not authenticate the integrity of the current message itself. 
 
 Discord
 ~~~~~~~

elastalert/alerters/dingtalk.py

@@ -1,4 +1,9 @@
+import base64
+import hashlib
+import hmac
 import json
+import time
+import urllib
 import warnings
 
 import requests
@@ -17,6 +22,7 @@ def __init__(self, rule):
         super(DingTalkAlerter, self).__init__(rule)
         self.dingtalk_access_token = self.rule.get('dingtalk_access_token', None)
         self.dingtalk_webhook_url = 'https://oapi.dingtalk.com/robot/send?access_token=%s' % (self.dingtalk_access_token)
+        self.dingtalk_sign = self.rule.get('dingtalk_sign', None)
         self.dingtalk_msgtype = self.rule.get('dingtalk_msgtype', 'text')
         self.dingtalk_single_title = self.rule.get('dingtalk_single_title', 'elastalert')
         self.dingtalk_single_url = self.rule.get('dingtalk_single_url', '')
@@ -26,6 +32,16 @@ def __init__(self, rule):
         self.dingtalk_proxy_login = self.rule.get('dingtalk_proxy_login', None)
         self.dingtalk_proxy_password = self.rule.get('dingtalk_proxy_pass', None)
 
+    def sign(self):
+        timestamp = str(round(time.time() * 1000))
+        secret = self.dingtalk_sign
+        secret_enc = secret.encode('utf-8')
+        string_to_sign = '{}\n{}'.format(timestamp, secret)
+        string_to_sign_enc = string_to_sign.encode('utf-8')
+        hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
+        sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
+        return timestamp, sign
+
     def alert(self, matches):
         title = self.create_title(matches)
         body = self.create_alert_body(matches)
@@ -36,7 +52,9 @@ def alert(self, matches):
             'Content-Type': 'application/json',
             'Accept': 'application/json;charset=utf-8'
         }
-
+        if self.dingtalk_sign:
+            timestamp, sign = self.sign()
+            self.dingtalk_webhook_url = '{}&timestamp={}&sign={}'.format(self.dingtalk_webhook_url, timestamp, sign)
         if self.dingtalk_msgtype == 'text':
             # text
             payload = {

elastalert/schema.yaml

@@ -438,6 +438,7 @@ properties:
   dingtalk_proxy: {type: string}
   dingtalk_proxy_login: {type: string}
   dingtalk_proxy_pass: {type: string}
+  dingtalk_sign: {type: string}
 
   ## Discord
   discord_webhook_url: {type: string}

tests/alerters/dingtalk_test.py

@@ -12,6 +12,71 @@
 from elastalert.util import EAException
 
 
+def test_dingtalk_sign(caplog):
+    caplog.set_level(logging.INFO)
+    rule = {
+        'type': 'any',
+        'dingtalk_sign': "xxx",
+        'alert': [],
+        'alert_subject': 'Test DingTalk'
+    }
+    rules_loader = FileRulesLoader({})
+    rules_loader.load_modules(rule)
+    alert = DingTalkAlerter(rule)
+
+    with mock.patch('time.time', return_value=1720063700033):
+        timestamp, sign = alert.sign()
+        assert sign == 'p7C%2BjdCVtGt77XXVBqP6I9ZzC8T0X9tv%2BWOduuQ53ZM%3D'
+
+
+def test_dingtalk_sign_text(caplog):
+    caplog.set_level(logging.INFO)
+    rule = {
+        'name': 'Test DingTalk Rule',
+        'type': 'any',
+        'dingtalk_access_token': 'xxxxxxx',
+        'dingtalk_msgtype': 'text',
+        'dingtalk_sign': "xxx",
+        'alert': [],
+        'alert_subject': 'Test DingTalk'
+    }
+    rules_loader = FileRulesLoader({})
+    rules_loader.load_modules(rule)
+    alert = DingTalkAlerter(rule)
+
+    match = {
+        '@timestamp': '2021-01-01T00:00:00',
+        'somefield': 'foobarbaz'
+    }
+    with mock.patch('requests.post') as mock_post_request:
+        with mock.patch('time.time', return_value=1720063700033):
+            timestamp, sign = alert.sign()
+            url = '{}&timestamp={}&sign={}'.format('https://oapi.dingtalk.com/robot/send?access_token=xxxxxxx', timestamp, sign)
+
+            alert.alert([match])
+
+            expected_data = {
+                'msgtype': 'text',
+                'text': {'content': 'Test DingTalk Rule\n\n@timestamp: 2021-01-01T00:00:00\nsomefield: foobarbaz\n'}
+            }
+
+            mock_post_request.assert_called_once_with(
+                url,
+                data=mock.ANY,
+                headers={
+                    'Content-Type': 'application/json',
+                    'Accept': 'application/json;charset=utf-8'
+                },
+                proxies=None,
+                auth=None
+            )
+
+            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 dingtalk') == caplog.record_tuples[0]
+
+
 def test_dingtalk_text(caplog):
     caplog.set_level(logging.INFO)
     rule = {