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 = {