alerta_alerta-contrib/plugins/msteams/alerta_msteams.py

import json
import logging
import os

import pymsteams
import requests
from alerta.plugins import PluginBase

try:
    from alerta.plugins import app  # alerta >= 5.0
except ImportError:
    from alerta.app import app  # alerta < 5.0

LOG = logging.getLogger('alerta.plugins.msteams')

try:
    from jinja2 import Template
except Exception as e:
    LOG.error(
        'MS Teams: ERROR - Jinja template error: %s, template functionality will be unavailable', e)

MS_TEAMS_COLORS_MAP = app.config.get('MS_TEAMS_COLORS_MAP', {})
MS_TEAMS_DEFAULT_COLORS_MAP = {'security': '000000',
                               'critical': 'D8122A',
                               'major': 'EA680F',
                               'minor': 'FFBE1E',
                               'warning': '1E90FF'}
MS_TEAMS_DEFAULT_COLOR = '00AA5A'
MS_TEAMS_DEFAULT_TIMEOUT = 7  # pymsteams http_timeout


class SendConnectorCardMessage(PluginBase):

    def __init__(self, name=None):
        # override user-defined severities(colors)
        self._colors = MS_TEAMS_DEFAULT_COLORS_MAP
        self._colors.update(MS_TEAMS_COLORS_MAP)

        super().__init__(name)

    def _load_template(self, templateFmt):
        try:
            if os.path.exists(templateFmt):
                with open(templateFmt) as f:
                    template = Template(f.read())
            else:
                template = Template(templateFmt)
            return template
        except Exception as e:
            LOG.error('MS Teams: ERROR - Template init failed: %s', e)
            return

    def pre_receive(self, alert, **kwargs):
        return alert

    def post_receive(self, alert, **kwargs):
        MS_TEAMS_WEBHOOK_URL = self.get_config(
            'MS_TEAMS_WEBHOOK_URL', default='', type=str, **kwargs)
        MS_TEAMS_SUMMARY_FMT = self.get_config(
            'MS_TEAMS_SUMMARY_FMT', default=None, type=str, **kwargs)  # Message summary(title) format
        MS_TEAMS_TEXT_FMT = self.get_config(
            'MS_TEAMS_TEXT_FMT', default=None, type=str, **kwargs)  # Message text format
        # json/Jinja2 MS teams messagecard payload
        MS_TEAMS_PAYLOAD = self.get_config(
            'MS_TEAMS_PAYLOAD', default=None, type=str, **kwargs)
        MS_TEAMS_INBOUNDWEBHOOK_URL = self.get_config(
            'MS_TEAMS_INBOUNDWEBHOOK_URL', default=None, type=str, **kwargs)  # webhook url for connectorcard actions
        # X-API-Key (needs webhook.write permission)
        MS_TEAMS_APIKEY = self.get_config(
            'MS_TEAMS_APIKEY', default=None, type=str, **kwargs)
        DASHBOARD_URL = self.get_config(
            'DASHBOARD_URL', default='', type=str, **kwargs)

        if alert.repeat:
            return

        color = self._colors.get(alert.severity, MS_TEAMS_DEFAULT_COLOR)
        url = '{}/#/alert/{}'.format(DASHBOARD_URL, alert.id)

        template_vars = {
            'alert': alert,
            'config': app.config,
            'color': color,
            'url': url
        }

        if MS_TEAMS_INBOUNDWEBHOOK_URL and MS_TEAMS_APIKEY:
            # Add X-API-Key header for teams(webhook) HttpPOST actions
            template_vars['headers'] = '[ {{ "name": "X-API-Key", "value": "{}" }} ]'.format(
                MS_TEAMS_APIKEY)
            template_vars['webhook_url'] = MS_TEAMS_INBOUNDWEBHOOK_URL

        if MS_TEAMS_PAYLOAD:
            # Use "raw" json ms teams message card format
            payload_template = self._load_template(MS_TEAMS_PAYLOAD)
            try:
                card_json = payload_template.render(**template_vars)
                LOG.debug('MS Teams payload(json): %s', card_json)
                # Try to check that we've valid json
                json.loads(card_json)
            except Exception as e:
                LOG.error(
                    'MS Teams: ERROR - Template(PAYLOAD) render failed: %s', e)
                return
        else:
            # Use pymsteams to format/construct message card
            if MS_TEAMS_SUMMARY_FMT:
                template = self._load_template(MS_TEAMS_SUMMARY_FMT)
                try:
                    summary = template.render(**template_vars)
                except Exception as e:
                    LOG.error('MS Teams: ERROR - Template render failed: %s', e)
                    return
            else:
                summary = ('<b>[{status}] {environment} {service} {severity} - <i>{event} on {resource}</i></b>').format(
                    status=alert.status.capitalize(),
                    environment=alert.environment.upper(),
                    service=','.join(alert.service),
                    severity=alert.severity.capitalize(),
                    event=alert.event,
                    resource=alert.resource
                )

            if MS_TEAMS_TEXT_FMT:
                txt_template = self._load_template(MS_TEAMS_TEXT_FMT)
                try:
                    text = txt_template.render(**template_vars)
                except Exception as e:
                    LOG.error(
                        'MS Teams: ERROR - Template(TEXT_FMT) render failed: %s', e)
                    return
            else:
                text = alert.text

            LOG.debug('MS Teams payload: %s', summary)

        try:
            if MS_TEAMS_PAYLOAD:
                # Use requests.post to send raw json message card
                LOG.debug('MS Teams sending(json payload) POST to %s',
                          MS_TEAMS_WEBHOOK_URL)
                r = requests.post(
                    MS_TEAMS_WEBHOOK_URL, data=card_json, timeout=MS_TEAMS_DEFAULT_TIMEOUT)
                LOG.debug('MS Teams response: %s / %s' %
                          (r.status_code, r.text))
            else:
                # Use pymsteams to send card
                msTeamsMessage = pymsteams.connectorcard(
                    hookurl=MS_TEAMS_WEBHOOK_URL, http_timeout=MS_TEAMS_DEFAULT_TIMEOUT)
                msTeamsMessage.title(summary)
                msTeamsMessage.text(text)
                msTeamsMessage.addLinkButton('Open in Alerta', url)
                msTeamsMessage.color(color)
                msTeamsMessage.send()
        except Exception as e:
            raise RuntimeError('MS Teams: ERROR - %s', e)

    def status_change(self, alert, status, text, **kwargs):
        return