Files
prowler/tests/lib/outputs/slack/slack_test.py

573 lines
20 KiB
Python

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,
AZURE_SUBSCRIPTION_NAME,
set_mocked_azure_provider,
)
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:
def test_create_message_identity_aws(self):
aws_provider = set_mocked_aws_provider()
slack = Slack(SLACK_TOKEN, SLACK_CHANNEL, aws_provider)
assert slack.__create_message_identity__(aws_provider) == (
f"AWS Account *{aws_provider.identity.account}*",
aws_logo,
)
def test_create_message_identity_azure(self):
azure_provider = set_mocked_azure_provider()
slack = Slack(SLACK_TOKEN, SLACK_CHANNEL, azure_provider)
assert slack.__create_message_identity__(azure_provider) == (
f"Azure Subscriptions:\n- *{AZURE_SUBSCRIPTION_ID}: {AZURE_SUBSCRIPTION_NAME}*\n",
azure_logo,
)
def test_create_message_identity_gcp(self):
gcp_provider = set_mocked_gcp_provider(
project_ids=["test-project1", "test-project2"],
)
slack = Slack(SLACK_TOKEN, SLACK_CHANNEL, gcp_provider)
assert slack.__create_message_identity__(gcp_provider) == (
f"GCP Projects *{', '.join(gcp_provider.project_ids)}*",
gcp_logo,
)
def test_create_title(self):
aws_provider = set_mocked_aws_provider()
slack = Slack(SLACK_TOKEN, SLACK_CHANNEL, aws_provider)
stats = {}
stats["total_pass"] = 12
stats["total_fail"] = 10
stats["total_critical_severity_pass"] = 4
stats["total_critical_severity_fail"] = 4
stats["total_high_severity_fail"] = 1
stats["total_high_severity_pass"] = 1
stats["total_medium_severity_fail"] = 2
stats["total_medium_severity_pass"] = 1
stats["total_low_severity_fail"] = 3
stats["total_low_severity_pass"] = 3
stats["resources_count"] = 20
stats["findings_count"] = 22
identity = slack.__create_message_identity__(aws_provider) == (
f"AWS Account *{aws_provider.identity.account}*",
aws_logo,
)
assert (
slack.__create_title__(identity, stats)
== f"Hey there 👋 \n I'm *Prowler*, _the handy multi-cloud security tool_ :cloud::key:\n\n I have just finished the security assessment on your {identity} with a total of *{stats['findings_count']}* findings."
)
def test_create_message_blocks_aws(self):
aws_provider = set_mocked_aws_provider()
slack = Slack(SLACK_TOKEN, SLACK_CHANNEL, aws_provider)
args = "--slack"
stats = {}
stats["total_pass"] = 12
stats["total_fail"] = 10
stats["total_critical_severity_pass"] = 2
stats["total_critical_severity_fail"] = 4
stats["total_high_severity_fail"] = 1
stats["total_high_severity_pass"] = 1
stats["total_medium_severity_fail"] = 2
stats["total_medium_severity_pass"] = 1
stats["total_low_severity_fail"] = 2
stats["total_low_severity_pass"] = 3
stats["resources_count"] = 20
stats["findings_count"] = 22
aws_identity = f"AWS Account *{AWS_ACCOUNT_NUMBER}*"
assert slack.__create_message_blocks__(aws_identity, aws_logo, stats, args) == [
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": slack.__create_title__(aws_identity, stats),
},
"accessory": {
"type": "image",
"image_url": aws_logo,
"alt_text": "Provider Logo",
},
},
{"type": "divider"},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": f"\n:white_check_mark: *{stats['total_pass']} Passed findings* ({round(stats['total_pass'] / stats['findings_count'] * 100 , 2)}%)\n",
},
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": (
"*Severities:*\n"
"• *Critical:* 2 "
"• *High:* 1 "
"• *Medium:* 1 "
"• *Low:* 3"
),
},
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": f"\n:x: *{stats['total_fail']} Failed findings* ({round(stats['total_fail'] / stats['findings_count'] * 100 , 2)}%)\n ",
},
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": (
"*Severities:*\n"
"• *Critical:* 4 "
"• *High:* 1 "
"• *Medium:* 2 "
"• *Low:* 2"
),
},
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": f"\n:bar_chart: *{stats['resources_count']} Scanned Resources*\n",
},
},
{"type": "divider"},
{
"type": "context",
"elements": [
{
"type": "mrkdwn",
"text": f"Used parameters: `prowler {args}`",
}
],
},
{"type": "divider"},
{
"type": "section",
"text": {"type": "mrkdwn", "text": "Join our Slack Community!"},
"accessory": {
"type": "button",
"text": {"type": "plain_text", "text": "Prowler :slack:"},
"url": "https://goto.prowler.com/slack",
},
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": "Feel free to contact us in our repo",
},
"accessory": {
"type": "button",
"text": {"type": "plain_text", "text": "Prowler :github:"},
"url": "https://github.com/prowler-cloud/prowler",
},
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": "See all the things you can do with ProwlerPro",
},
"accessory": {
"type": "button",
"text": {"type": "plain_text", "text": "Prowler Pro"},
"url": "https://prowler.pro",
},
},
]
def test_create_message_blocks_azure(self):
aws_provider = set_mocked_azure_provider()
slack = Slack(SLACK_TOKEN, SLACK_CHANNEL, aws_provider)
args = "--slack"
stats = {}
stats["total_pass"] = 12
stats["total_fail"] = 10
stats["total_critical_severity_pass"] = 2
stats["total_critical_severity_fail"] = 4
stats["total_high_severity_fail"] = 1
stats["total_high_severity_pass"] = 1
stats["total_medium_severity_fail"] = 2
stats["total_medium_severity_pass"] = 1
stats["total_low_severity_fail"] = 2
stats["total_low_severity_pass"] = 3
stats["resources_count"] = 20
stats["findings_count"] = 22
azure_identity = "Azure Subscriptions:\n- *subscription 1: qwerty*\n- *subscription 2: asdfg*\n"
assert slack.__create_message_blocks__(
azure_identity, azure_logo, stats, args
) == [
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": slack.__create_title__(azure_identity, stats),
},
"accessory": {
"type": "image",
"image_url": azure_logo,
"alt_text": "Provider Logo",
},
},
{"type": "divider"},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": f"\n:white_check_mark: *{stats['total_pass']} Passed findings* ({round(stats['total_pass'] / stats['findings_count'] * 100 , 2)}%)\n",
},
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": (
"*Severities:*\n"
"• *Critical:* 2 "
"• *High:* 1 "
"• *Medium:* 1 "
"• *Low:* 3"
),
},
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": f"\n:x: *{stats['total_fail']} Failed findings* ({round(stats['total_fail'] / stats['findings_count'] * 100 , 2)}%)\n ",
},
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": (
"*Severities:*\n"
"• *Critical:* 4 "
"• *High:* 1 "
"• *Medium:* 2 "
"• *Low:* 2"
),
},
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": f"\n:bar_chart: *{stats['resources_count']} Scanned Resources*\n",
},
},
{"type": "divider"},
{
"type": "context",
"elements": [
{
"type": "mrkdwn",
"text": f"Used parameters: `prowler {args}`",
}
],
},
{"type": "divider"},
{
"type": "section",
"text": {"type": "mrkdwn", "text": "Join our Slack Community!"},
"accessory": {
"type": "button",
"text": {"type": "plain_text", "text": "Prowler :slack:"},
"url": "https://goto.prowler.com/slack",
},
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": "Feel free to contact us in our repo",
},
"accessory": {
"type": "button",
"text": {"type": "plain_text", "text": "Prowler :github:"},
"url": "https://github.com/prowler-cloud/prowler",
},
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": "See all the things you can do with ProwlerPro",
},
"accessory": {
"type": "button",
"text": {"type": "plain_text", "text": "Prowler Pro"},
"url": "https://prowler.pro",
},
},
]
def test_create_message_blocks_gcp(self):
aws_provider = set_mocked_gcp_provider()
slack = Slack(SLACK_TOKEN, SLACK_CHANNEL, aws_provider)
args = "--slack"
stats = {}
stats["total_pass"] = 12
stats["total_fail"] = 10
stats["total_critical_severity_pass"] = 2
stats["total_critical_severity_fail"] = 4
stats["total_high_severity_fail"] = 1
stats["total_high_severity_pass"] = 1
stats["total_medium_severity_fail"] = 2
stats["total_medium_severity_pass"] = 1
stats["total_low_severity_fail"] = 2
stats["total_low_severity_pass"] = 3
stats["resources_count"] = 20
stats["findings_count"] = 22
gcp_identity = "GCP Project *gcp-project*"
assert slack.__create_message_blocks__(gcp_identity, gcp_logo, stats, args) == [
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": slack.__create_title__(gcp_identity, stats),
},
"accessory": {
"type": "image",
"image_url": gcp_logo,
"alt_text": "Provider Logo",
},
},
{"type": "divider"},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": f"\n:white_check_mark: *{stats['total_pass']} Passed findings* ({round(stats['total_pass'] / stats['findings_count'] * 100 , 2)}%)\n",
},
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": (
"*Severities:*\n"
"• *Critical:* 2 "
"• *High:* 1 "
"• *Medium:* 1 "
"• *Low:* 3"
),
},
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": f"\n:x: *{stats['total_fail']} Failed findings* ({round(stats['total_fail'] / stats['findings_count'] * 100 , 2)}%)\n ",
},
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": (
"*Severities:*\n"
"• *Critical:* 4 "
"• *High:* 1 "
"• *Medium:* 2 "
"• *Low:* 2"
),
},
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": f"\n:bar_chart: *{stats['resources_count']} Scanned Resources*\n",
},
},
{"type": "divider"},
{
"type": "context",
"elements": [
{
"type": "mrkdwn",
"text": f"Used parameters: `prowler {args}`",
}
],
},
{"type": "divider"},
{
"type": "section",
"text": {"type": "mrkdwn", "text": "Join our Slack Community!"},
"accessory": {
"type": "button",
"text": {"type": "plain_text", "text": "Prowler :slack:"},
"url": "https://goto.prowler.com/slack",
},
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": "Feel free to contact us in our repo",
},
"accessory": {
"type": "button",
"text": {"type": "plain_text", "text": "Prowler :github:"},
"url": "https://github.com/prowler-cloud/prowler",
},
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": "See all the things you can do with ProwlerPro",
},
"accessory": {
"type": "button",
"text": {"type": "plain_text", "text": "Prowler Pro"},
"url": "https://prowler.pro",
},
},
]
def test_send_slack_message(self):
mocked_slack_response = {
"ok": True,
"channel": "XXXXXXXXXX",
"ts": "1683623300.083429",
"message": {
"type": "message",
"subtype": "bot_message",
"text": "",
"ts": "1683623300.083429",
"username": "Prowler",
"icons": {},
"bot_id": "B055L25CVFH",
"app_id": "A055U03H2QN",
"blocks": [],
},
}
mocked_web_client = mock.MagicMock
mocked_web_client.chat_postMessage = mock.Mock(
return_value=mocked_slack_response
)
with mock.patch(
"prowler.lib.outputs.slack.slack.WebClient", new=mocked_web_client
):
aws_provider = set_mocked_aws_provider()
slack = Slack(SLACK_TOKEN, SLACK_CHANNEL, aws_provider)
stats = {}
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)