Dynamic Slack Webhook Option

[VirusProtect]

Hi, is it possible to create an enhancement in ElastAlert2 where the slack_webhook_url changes based on the value of the source.ip field? For example:

If source.ip contains the value 8.8.8.8, use slack_webhook_url: 'xxxxxxx'.
If source.ip contains the value 4.4.4.4, use slack_webhook_url: 'yyyyyyyy'.

I’ve tried a few scripts, but the alert only gets sent to the webhook URL specified in the main rule configuration.

Or is there any other way to achieve this?

[jertel]

I would think this is possible. Show us the enchancement code you tried; perhaps someone will spot a problem.

[VirusProtect]

I tried a couple of enhancement codes:

The first code should add an extra webhook to send alerts to multiple Slack channels if data_stream.namespace is value is "test":

from elastalert.enhancements import BaseEnhancement

class Enhancement(BaseEnhancement):
    def process(self, match):
        if 'data_stream.namespace' in match:
            namespace = match['data_stream.namespace']
            
            if namespace == 'test':
                extra_webhook = 'https://hooks.slack.com/xxxxxxxx'
                
                if isinstance(self.rule['slack_webhook_url'], list):
                    self.rule['slack_webhook_url'].append(extra_webhook)
                else:
                    self.rule['slack_webhook_url'] = [self.rule['slack_webhook_url'], extra_webhook]

The second enhancement I tried is to replace an already existing webhook:

from elastalert.enhancements import BaseEnhancement

class ReplacementEnhancement(BaseEnhancement):
    def process(self, match):
        if 'data_stream.namespace' in match:
            namespace = match['data_stream.namespace']
            
            if namespace == 'test':
                new_webhook = 'https://hooks.slack.com/yyyyy'
                
                if isinstance(self.rule['slack_webhook_url'], list):
                    self.rule['slack_webhook_url'] = [
                        new_webhook if url == 'https://hooks.slack.com/xxxxxxxx' else url
                        for url in self.rule['slack_webhook_url']
                    ]
                else:
                    if self.rule['slack_webhook_url'] == 'https://hooks.slack.com/xxxxxxxx':
                        self.rule['slack_webhook_url'] = new_webhook

[jertel]

Slack alerter URLs are initialized when ElastAlert 2 starts up and loads the rule into memory. At that time the Alerter class is initialized with the configured URL. So changing the rule object's slack URL at runtime won't make a difference since the alerter instance was already initialized.

The Slack alerter supports dynamically changing rooms/channels, but not the actual Slack webhook itself.

To do what you're attempting would require manipulating the URL in the alerter instance, not the rule instance. Ex:

        for alerter in self.rule['alert']:
            if isinstance(alerter, SlackAlerter):
                alerter.slack_webhook_url = [new_webhook]

Also, the slack_webhook_url is always converted to a list, so don't try assigned a raw string to that property.

Finally, doing this is risky because the next alert will be using your new URL, even if it didn't match your conditions. Your enhancement would need to always set the desired URL and never rely on what was provided in the rule yaml file.

[VirusProtect]

I tried this one, but it is still not working :

from elastalert.enhancements import BaseEnhancement
from elastalert.alerts import SlackAlerter

class Enhancement(BaseEnhancement):
    def process(self, match):
        namespace = match.get('data_stream', {}).get('namespace')

        if namespace == 'test':
            new_webhook = 'https://hooks.slack.com/xxx'
        elif namespace == 'orange':
            new_webhook = 'https://hooks.slack.com/yyy'
        else:
            new_webhook = 'https://hooks.slack.com/zzz'

        for alerter in self.rule['alert']:
            if isinstance(alerter, SlackAlerter):
                alerter.slack_webhook_url = [new_webhook]

[jertel]

I tried it before I sent my reply and it did work. If it's not working for you then I suggest adding debugging print lines to step through the logic.