Add custom User-Agent header support for Mattermost integration

docs/configuration/sinks/mattermost.rst

@@ -75,6 +75,7 @@ Now we're ready to configure the Mattermost sink.
             token_id: <YOUR BOT TOKEN ID> (the token id visible in bot panel)
             channel: <YOUR CHANNEL NAME> (the channel name you want to send messages to - either display name or channel name divided by hyphen (e.g. channel-name))
             team_id: <YOUR TEAM ID> (OPTIONAL - this is only needed if your mattermost bot is not an admin)
+            user_agent: <YOUR USER AGENT> (OPTIONAL - custom User-Agent header for Mattermost API requests)
 
 Save the file and run
 

src/robusta/core/sinks/mattermost/mattermost_sink.py

@@ -16,10 +16,13 @@ def __init__(self, sink_config: MattermostSinkConfigWrapper, registry):
             token_id=sink_config.mattermost_sink.token_id,
             team=sink_config.mattermost_sink.team,
             team_id=sink_config.mattermost_sink.team_id,
+            user_agent=sink_config.mattermost_sink.user_agent,
         )
         self.sender = MattermostSender(
-            cluster_name=self.cluster_name, account_id=self.account_id, client=client,
-            sink_params=sink_config.mattermost_sink
+            cluster_name=self.cluster_name,
+            account_id=self.account_id,
+            client=client,
+            sink_params=sink_config.mattermost_sink,
         )
 
     def write_finding(self, finding: Finding, platform_enabled: bool):

src/robusta/core/sinks/mattermost/mattermost_sink_params.py

@@ -14,6 +14,7 @@ class MattermostSinkParams(SinkBaseParams):
     channel: str
     team: Optional[str]
     team_id: Optional[str]
+    user_agent: Optional[str]
 
     @classmethod
     def _get_sink_type(cls):

src/robusta/integrations/mattermost/client.py

@@ -6,25 +6,39 @@
 
 _API_PREFIX = "api/v4"
 
+
 class MattermostClient:
     channel_id: str
     bot_id: str
     team_id: Optional[str]
-
-    def __init__(self, url: str, token: str, token_id: str, channel_name: str, team: Optional[str], team_id: Optional[str]):
+    user_agent: Optional[str]
+
+    def __init__(
+        self,
+        url: str,
+        token: str,
+        token_id: str,
+        channel_name: str,
+        team: Optional[str],
+        team_id: Optional[str],
+        user_agent: Optional[str],
+    ):
         """
         Set the Mattermost webhook url.
         """
         self.client_url = url
         self.token = token
         self.token_id = token_id
         self.team_id = team_id
+        self.user_agent = user_agent
         self.is_admin = self.is_admin_bot()
         self._init_setup(channel_name, team)
 
     def _send_mattermost_request(self, url: str, method: HttpMethod, **kwargs):
         headers = kwargs.pop("headers", {})
         headers["Authorization"] = f"Bearer {self.token}"
+        if self.user_agent:
+            headers["User-Agent"] = self.user_agent
         return process_request(url, method, headers=headers, **kwargs)
 
     def _get_full_mattermost_url(self, endpoint: str) -> str: