feat(slack): add test_connection method (#5340)

prowler/lib/outputs/slack/exceptions/exceptions.py

@@ -0,0 +1,61 @@
+from prowler.exceptions.exceptions import ProwlerException
+
+
+# Exceptions codes from 8000 to 8999 are reserved for Slack exceptions
+class SlackBaseException(ProwlerException):
+    """Base class for Slack errors."""
+
+    SLACK_ERROR_CODES = {
+        (8000, "SlackClientError"): {
+            "message": "Slack ClientError occurred",
+            "remediation": "Check your Slack client configuration and permissions.",
+        },
+        (8001, "SlackNoCredentialsError"): {
+            "message": "Invalid Slack credentials found",
+            "remediation": "Some aspect of authentication cannot be validated. Either the provided token is invalid or the request originates from an IP address disallowed from making the request.",
+        },
+        (8002, "SlackChannelNotFound"): {
+            "message": "Slack channel not found",
+            "remediation": "Check the channel name and ensure it exists.",
+        },
+    }
+
+    def __init__(self, code, file=None, original_exception=None, message=None):
+        error_info = self.SLACK_ERROR_CODES.get((code, self.__class__.__name__))
+        if message:
+            error_info["message"] = message
+        super().__init__(
+            code,
+            source="Slack",
+            file=file,
+            original_exception=original_exception,
+            error_info=error_info,
+        )
+
+
+class SlackCredentialsError(SlackBaseException):
+    """Base class for Slack credentials errors."""
+
+    def __init__(self, code, file=None, original_exception=None, message=None):
+        super().__init__(code, file, original_exception, message)
+
+
+class SlackClientError(SlackCredentialsError):
+    def __init__(self, file=None, original_exception=None, message=None):
+        super().__init__(
+            8000, file=file, original_exception=original_exception, message=message
+        )
+
+
+class SlackNoCredentialsError(SlackCredentialsError):
+    def __init__(self, file=None, original_exception=None, message=None):
+        super().__init__(
+            8001, file=file, original_exception=original_exception, message=message
+        )
+
+
+class SlackChannelNotFound(SlackCredentialsError):
+    def __init__(self, file=None, original_exception=None, message=None):
+        super().__init__(
+            8002, file=file, original_exception=original_exception, message=message
+        )

prowler/lib/outputs/slack/slack.py

@@ -1,3 +1,4 @@
+import os
 from typing import Any
 
 from slack_sdk import WebClient
@@ -5,6 +6,12 @@ from slack_sdk.web.base_client import SlackResponse
 
 from prowler.config.config import aws_logo, azure_logo, gcp_logo, square_logo_img
 from prowler.lib.logger import logger
+from prowler.lib.outputs.slack.exceptions.exceptions import (
+    SlackChannelNotFound,
+    SlackClientError,
+    SlackNoCredentialsError,
+)
+from prowler.providers.common.models import Connection
 
 
 class Slack:
@@ -51,7 +58,6 @@ class Slack:
             logger.error(
                 f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
             )
-            return error
 
     def __create_message_identity__(self, provider: Any):
         """
@@ -231,3 +237,66 @@ class Slack:
             logger.error(
                 f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
             )
+
+    @staticmethod
+    def test_connection(
+        token: str,
+        channel: str,
+        raise_on_exception: bool = True,
+    ) -> Connection:
+        """
+        Test the Slack connection by validating the provided token and channel.
+
+        Args:
+            token (str): The Slack token to be tested.
+            channel (str): The Slack channel to be validated.
+
+        Returns:
+            Connection: A Connection object.
+        """
+        try:
+            client = WebClient(token=token)
+            # Test if the token is valid
+            auth_response = client.auth_test()
+            if auth_response["ok"]:
+                # Test if the channel is accessible
+                channels_response = client.conversations_info(
+                    token=token, channel=channel
+                )
+                if channels_response["ok"]:
+                    return Connection(is_connected=True)
+                else:
+                    exception = SlackChannelNotFound(
+                        file=os.path.basename(__file__),
+                        message=(
+                            channels_response["error"]
+                            if "error" in channels_response
+                            else "Unknown error"
+                        ),
+                    )
+                    if raise_on_exception:
+                        raise exception
+                    return Connection(error=exception)
+            else:
+                exception = SlackNoCredentialsError(
+                    file=os.path.basename(__file__),
+                    message=(
+                        auth_response["error"]
+                        if "error" in auth_response
+                        else "Unknown error"
+                    ),
+                )
+                if raise_on_exception:
+                    raise exception
+                return Connection(error=exception)
+
+        except Exception as error:
+            logger.error(
+                f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+            if raise_on_exception:
+                raise SlackClientError(
+                    file=os.path.basename(__file__),
+                    original_exception=error,
+                ) from error
+            return Connection(error=error)

tests/lib/outputs/slack/slack_test.py

@@ -1,7 +1,13 @@
 from unittest import mock
 
 from prowler.config.config import aws_logo, azure_logo, gcp_logo
+from prowler.lib.outputs.slack.exceptions.exceptions import (
+    SlackChannelNotFound,
+    SlackClientError,
+    SlackNoCredentialsError,
+)
 from prowler.lib.outputs.slack.slack import Slack
+from prowler.providers.common.models import Connection
 from tests.providers.aws.utils import AWS_ACCOUNT_NUMBER, set_mocked_aws_provider
 from tests.providers.azure.azure_fixtures import (
     AZURE_SUBSCRIPTION_ID,
@@ -12,6 +18,7 @@ from tests.providers.gcp.gcp_fixtures import set_mocked_gcp_provider
 
 SLACK_CHANNEL = "test-channel"
 SLACK_TOKEN = "test-token"
+NON_EXISTING_CHANNEL = "non-existing-channel"
 
 
 class TestSlackIntegration:
@@ -486,3 +493,80 @@ class TestSlackIntegration:
             args = "--slack"
             response = slack.send(stats, args)
             assert response == mocked_slack_response
+
+    def test_test_connection(self):
+        mocked_auth_response = {"ok": True}
+        mocked_conversations_info = {
+            "ok": True,
+            "channels": [
+                {"id": "C87654321", "name": SLACK_CHANNEL, "is_member": True},
+            ],
+        }
+        mocked_web_client = mock.MagicMock()
+        mocked_web_client.auth_test = mock.Mock(return_value=mocked_auth_response)
+        mocked_web_client.conversations_info = mock.Mock(
+            return_value=mocked_conversations_info
+        )
+        with mock.patch(
+            "prowler.lib.outputs.slack.slack.WebClient", return_value=mocked_web_client
+        ):
+            assert Slack.test_connection(
+                token=SLACK_TOKEN, channel=SLACK_CHANNEL
+            ) == Connection(is_connected=True)
+
+    def test_slack_no_credentials_error(self):
+        mocked_auth_response = {"ok": False, "error": "invalid_auth"}
+        mocked_web_client = mock.MagicMock()
+        mocked_web_client.auth_test = mock.Mock(return_value=mocked_auth_response)
+
+        with mock.patch(
+            "prowler.lib.outputs.slack.slack.WebClient", return_value=mocked_web_client
+        ):
+            connection = Slack.test_connection(
+                token=SLACK_TOKEN,
+                channel=NON_EXISTING_CHANNEL,
+                raise_on_exception=False,
+            )
+
+            assert not connection.is_connected
+            assert isinstance(connection.error, SlackNoCredentialsError)
+            assert "invalid_auth" in str(connection.error)
+
+    def test_slack_channel_not_found(self):
+        mocked_auth_response = {"ok": True}
+        mocked_conversations_info = {"ok": False, "error": "channel_not_found"}
+        mocked_web_client = mock.MagicMock()
+        mocked_web_client.auth_test = mock.Mock(return_value=mocked_auth_response)
+        mocked_web_client.conversations_info = mock.Mock(
+            return_value=mocked_conversations_info
+        )
+
+        with mock.patch(
+            "prowler.lib.outputs.slack.slack.WebClient", return_value=mocked_web_client
+        ):
+            connection = Slack.test_connection(
+                token=SLACK_TOKEN,
+                channel=NON_EXISTING_CHANNEL,
+                raise_on_exception=False,
+            )
+
+            assert not connection.is_connected
+            assert isinstance(connection.error, SlackChannelNotFound)
+            assert "channel_not_found" in str(connection.error)
+
+    def test_slack_client_error(self):
+        mocked_web_client = mock.MagicMock()
+        mocked_web_client.auth_test = mock.Mock(side_effect=SlackClientError)
+
+        with mock.patch(
+            "prowler.lib.outputs.slack.slack.WebClient", return_value=mocked_web_client
+        ):
+            connection = Slack.test_connection(
+                token=SLACK_TOKEN,
+                channel=NON_EXISTING_CHANNEL,
+                raise_on_exception=False,
+            )
+
+            assert not connection.is_connected
+            assert isinstance(connection.error, SlackClientError)
+            assert "Slack ClientError occurred" in str(connection.error)