Added support for webhook_url dynamic overriding using annotations (issue 1083) (#1476)

* Created MsTeamsWebhookUrlTransformer to enable overriding the webhook_url using annotations from workload or from environment variable

Author: itisallgood

docs/configuration/sinks/ms-teams.rst

@@ -20,6 +20,7 @@ Configuring the MS Teams sink
         - ms_teams_sink:
             name: main_ms_teams_sink
             webhook_url: teams-incoming-webhook  # see instructions below
+            webhook_override: DYNAMIC MS TEAMS WEBHOOK URL OVERRIDE (Optional)
 
 Then do a :ref:`Helm Upgrade <Simple Upgrade>`.
 
@@ -35,3 +36,40 @@ Obtaining a webhook URL
 .. image:: /images/msteams_sink/msteam_get_webhook_url.gif
     :width: 1024
     :align: center
+
+
+Dynamically Route MS Teams Alerts
+-------------------------------------------------------------------
+
+You can set the MS Teams webhook url value dynamically, based on the value of a specific ``annotation`` and environmental variable passed to runner.
+
+This can be done using the optional ``webhook_override`` sink parameter.
+
+As for now, the ``webhook_override`` parameter supports retrieving values specifically from annotations. You can specify an annotation key to retrieve the MS Teams webhook URL using the format ``annotations.<annotation_key>``. For example, if you use ``annotations.ms-team-alerts-sink``, the webhook URL will be taken from an annotation with the key ``ms-team-alerts-sink``.
+
+If the specified annotation does not exist, the default webhook URL from the ``webhook_url`` parameter will be used. If the annotation exists but does not contain a URL, the system will look for an environmental variable with the name matching the ``annotation`` value.
+
+.. code-block:: yaml
+
+     sinksConfig:
+     # MS Teams integration params
+    - ms_teams_sink:
+        name: main_ms_teams_sink
+        webhook_url: teams-incoming-webhook  # see instructions below
+        webhook_override: "annotations.ms-team-alerts-sink"
+
+A replacement pattern is also allowed, using ``$`` sign, before the variable.
+For cases where labels or annotations include special characters, such as ``${annotations.kubernetes.io/service-name}``, you can use the `${}` replacement pattern to represent the entire key, including special characters.
+For example, if you want to dynamically set the MS Teams webhook url based on the annotation ``kubernetes.io/service-name``, you can use the following syntax:
+
+- ``webhook_override: "${annotations.kubernetes.io/service-name}"``
+
+Example:
+
+.. code-block:: yaml
+
+        sinksConfig:
+        - ms_teams_sink:
+            name: main_ms_teams_sink
+            webhook_url: teams-incoming-webhook  # see instructions below
+            webhook_override: ${annotations.kubernetes.io/service-name}

src/robusta/core/sinks/common/channel_transformer.py

@@ -1,6 +1,6 @@
 from collections import defaultdict
 from string import Template
-from typing import Dict, Optional, Union
+from typing import Dict, Optional
 
 import regex
 
@@ -18,21 +18,7 @@
 MISSING = "<missing>"
 
 
-class ChannelTransformer:
-    @classmethod
-    def validate_channel_override(cls, v: Union[str, None]):
-        if v:
-            if regex.match(ONLY_VALUE_PATTERN, v):
-                return "$" + v
-            if not regex.match(COMPOSITE_PATTERN, v):
-                err_msg = (
-                    f"channel_override must be '{CLUSTER_PREF}' or '{LABELS_PREF}foo' or '{ANNOTATIONS_PREF}foo' "
-                    f"or contain patters like: '${CLUSTER_PREF}'/'${LABELS_PREF}foo'/"
-                    f"'${ANNOTATIONS_PREF}foo'"
-                )
-                raise ValueError(err_msg)
-        return v
-
+class BaseChannelTransformer:
     @classmethod
     def normalize_key_string(cls, s: str) -> str:
         return s.replace("/", "_").replace(".", "_").replace("-", "_")
@@ -47,7 +33,7 @@ def normalize_dict_keys(cls, metadata: Dict) -> Dict:
     # else, if found, return replacement else return MISSING
     @classmethod
     def get_replacement(cls, prefix: str, value: str, normalized_replacements: Dict) -> str:
-        if prefix in value:  # value is in the format of "$prefix" or "prefix"
+        if prefix in value:
             value = cls.normalize_key_string(value.replace(prefix, ""))
             if "$" in value:
                 return Template(value).safe_substitute(normalized_replacements)
@@ -56,13 +42,7 @@ def get_replacement(cls, prefix: str, value: str, normalized_replacements: Dict)
         return ""
 
     @classmethod
-    def replace_token(
-        cls,
-        pattern: regex.Pattern,
-        prefix: str,
-        channel: str,
-        replacements: Dict[str, str],
-    ) -> str:
+    def replace_token(cls, pattern: regex.Pattern, prefix: str, channel: str, replacements: Dict[str, str]) -> str:
         tokens = pattern.findall(channel)
         for token in tokens:
             clean_token = token.replace("{", "").replace("}", "")
@@ -71,6 +51,30 @@ def replace_token(
                 channel = channel.replace(token, replacement)
         return channel
 
+    @classmethod
+    def process_template_annotations(cls, channel: str, annotations: Dict[str, str]) -> str:
+        if ANNOTATIONS_PREF in channel:
+            normalized_annotations = cls.normalize_dict_keys(annotations)
+            channel = cls.replace_token(BRACKETS_PATTERN, ANNOTATIONS_PREF, channel, normalized_annotations)
+            channel = cls.replace_token(ANNOTATIONS_PREF_PATTERN, ANNOTATIONS_PREF, channel, normalized_annotations)
+        return channel
+
+
+class ChannelTransformer(BaseChannelTransformer):
+    @classmethod
+    def validate_channel_override(cls, v: Optional[str]) -> str:
+        if v:
+            if regex.match(ONLY_VALUE_PATTERN, v):
+                return "$" + v
+            if not regex.match(COMPOSITE_PATTERN, v):
+                err_msg = (
+                    f"channel_override must be '{CLUSTER_PREF}' or '{LABELS_PREF}foo' or '{ANNOTATIONS_PREF}foo' "
+                    f"or contain patters like: '${CLUSTER_PREF}'/'${LABELS_PREF}foo'/"
+                    f"'${ANNOTATIONS_PREF}foo'"
+                )
+                raise ValueError(err_msg)
+        return v
+
     @classmethod
     def template(
         cls,
@@ -93,14 +97,6 @@ def template(
             channel = cls.replace_token(BRACKETS_PATTERN, LABELS_PREF, channel, normalized_labels)
             channel = cls.replace_token(LABEL_PREF_PATTERN, LABELS_PREF, channel, normalized_labels)
 
-        if ANNOTATIONS_PREF in channel:
-            normalized_annotations = cls.normalize_dict_keys(annotations)
-            channel = cls.replace_token(BRACKETS_PATTERN, ANNOTATIONS_PREF, channel, normalized_annotations)
-            channel = cls.replace_token(
-                ANNOTATIONS_PREF_PATTERN,
-                ANNOTATIONS_PREF,
-                channel,
-                normalized_annotations,
-            )
+        channel = cls.process_template_annotations(channel, annotations)
 
         return channel if MISSING not in channel else default_channel

src/robusta/core/sinks/msteams/msteams_sink.py

@@ -8,8 +8,9 @@ class MsTeamsSink(SinkBase):
     def __init__(self, sink_config: MsTeamsSinkConfigWrapper, registry):
         super().__init__(sink_config.ms_teams_sink, registry)
         self.webhook_url = sink_config.ms_teams_sink.webhook_url
+        self.webhook_override = sink_config.ms_teams_sink.webhook_override
 
     def write_finding(self, finding: Finding, platform_enabled: bool):
         MsTeamsSender.send_finding_to_ms_teams(
-            self.webhook_url, finding, platform_enabled, self.cluster_name, self.account_id
+            self.webhook_url, finding, platform_enabled, self.cluster_name, self.account_id, self.webhook_override
         )

src/robusta/core/sinks/msteams/msteams_sink_params.py

@@ -1,14 +1,24 @@
+from typing import Optional
+
+from pydantic import validator
+
+from robusta.core.sinks.msteams.msteams_webhook_tranformer import MsTeamsWebhookUrlTransformer
 from robusta.core.sinks.sink_base_params import SinkBaseParams
 from robusta.core.sinks.sink_config import SinkConfigBase
 
 
 class MsTeamsSinkParams(SinkBaseParams):
     webhook_url: str
+    webhook_override: Optional[str] = None
 
     @classmethod
     def _get_sink_type(cls):
         return "msteams"
 
+    @validator("webhook_override")
+    def validate_webhook_override(cls, v: str):
+        return MsTeamsWebhookUrlTransformer.validate_webhook_override(v)
+
 
 class MsTeamsSinkConfigWrapper(SinkConfigBase):
     ms_teams_sink: MsTeamsSinkParams

src/robusta/core/sinks/msteams/msteams_webhook_tranformer.py

@@ -0,0 +1,69 @@
+import logging
+import os
+from typing import Dict, Optional
+
+import regex
+
+from robusta.core.sinks.common.channel_transformer import ANNOTATIONS_PREF, MISSING, BaseChannelTransformer
+
+ANNOTATIONS_COMPOSITE_PATTERN = r".*\$({?annotations.[^$]+).*"
+ANNOTATIONS_ONLY_VALUE_PATTERN = r"^(annotations.[^$]+)$"
+URL_PATTERN = regex.compile(
+    r"^(https?)://"
+    r"(?:(?:[A-Z0-9](?:[A-Z0-9-]{0,61}[A-Z0-9])?\.)+(?:[A-Z]{2,6}\.?|[A-Z0-9-]{2,}\.?)|"
+    r"localhost|"
+    r"\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}|"
+    r"\[?[A-F0-9]*:[A-F0-9:]+\]?)"
+    r"(?::\d+)?"
+    r"(?:/?|[/?]\S+)$",
+    regex.IGNORECASE,
+)
+
+
+# This class supports overriding the webhook_url only using annotations from yaml files.
+# Annotations are used instead of labels because urls can be passed to annotations contrary to labels.
+# Labels must be an empty string or consist of alphanumeric characters, '-', '_', or '.',
+# and must start and end with an alphanumeric character (e.g., 'MyValue', 'my_value', or '12345').
+# The regex used for label validation is '(([A-Za-z0-9][-A-Za-z0-9_.]*)?[A-Za-z0-9])?'.
+class MsTeamsWebhookUrlTransformer(BaseChannelTransformer):
+    @classmethod
+    def validate_webhook_override(cls, v: Optional[str]) -> Optional[str]:
+        if v:
+            if regex.match(ANNOTATIONS_ONLY_VALUE_PATTERN, v):
+                return "$" + v
+            if not regex.match(ANNOTATIONS_COMPOSITE_PATTERN, v):
+                err_msg = f"webhook_override must be '{ANNOTATIONS_PREF}foo' or contain patterns like: '${ANNOTATIONS_PREF}foo'"
+                raise ValueError(err_msg)
+        return v
+
+    @classmethod
+    def validate_url_or_get_env(cls, webhook_url: str, default_webhook_url: str) -> str:
+        if URL_PATTERN.match(webhook_url):
+            return webhook_url
+        logging.info(f"URL matching failed for: {webhook_url}. Trying to get environment variable.")
+
+        env_value = os.getenv(webhook_url)
+        if env_value:
+            return env_value
+        logging.info(f"Environment variable not found for: {webhook_url}. Using default webhook URL.")
+
+        return default_webhook_url
+
+    @classmethod
+    def template(
+        cls,
+        webhook_override: Optional[str],
+        default_webhook_url: str,
+        annotations: Dict[str, str],
+    ) -> str:
+        if not webhook_override:
+            return default_webhook_url
+
+        webhook_url = webhook_override
+
+        webhook_url = cls.process_template_annotations(webhook_url, annotations)
+        if MISSING in webhook_url:
+            return default_webhook_url
+        webhook_url = cls.validate_url_or_get_env(webhook_url, default_webhook_url)
+
+        return webhook_url

src/robusta/integrations/msteams/sender.py

@@ -13,6 +13,7 @@
     MarkdownBlock,
     TableBlock,
 )
+from robusta.core.sinks.msteams.msteams_webhook_tranformer import MsTeamsWebhookUrlTransformer
 from robusta.integrations.msteams.msteams_msg import MsTeamsMsg
 
 
@@ -50,8 +51,17 @@ def __split_block_to_files_and_all_the_rest(cls, enrichment: Enrichment):
 
     @classmethod
     def send_finding_to_ms_teams(
-        cls, webhook_url: str, finding: Finding, platform_enabled: bool, cluster_name: str, account_id: str
+        cls,
+        webhook_url: str,
+        finding: Finding,
+        platform_enabled: bool,
+        cluster_name: str,
+        account_id: str,
+        webhook_override: str,
     ):
+        webhook_url = MsTeamsWebhookUrlTransformer.template(
+            webhook_override=webhook_override, default_webhook_url=webhook_url, annotations=finding.subject.annotations
+        )
         msg = MsTeamsMsg(webhook_url)
         msg.write_title_and_desc(platform_enabled, finding, cluster_name, account_id)