DSPMCreateRiskSlackBlocks

Source

import demistomock as demisto  # noqa: F401
from CommonServerPython import *  # noqa: F401
from typing import Any


def create_slack_block(incident: dict, rule_names_dict: dict, incidentLink: str) -> dict:
    """
    Creates a Slack block message structure for a DSPM incident.

    The Slack block contains incident details such as Incident ID, DSPM Risk ID, Rule Name,
    Severity, Asset Information, and Remediation instructions. Additionally, a radio button
    to 'Create a Jira ticket' is included, and if the rule name matches specific conditions,
    an option to 'Remediate a Risk' is added.

    Args:
        incident (dict): A dictionary containing details of the DSPM incident.
        rule_names_dict (dict): A dictonary of rule names that trigger the addition of a 'Remediate a Risk' option.

    Returns:
        block (dict): A structured Slack block message in JSON format to be sent to Slack.
    """
    rule_name = incident.get("ruleName")

    # Slack block structure
    block: dict[str, list[dict[str, Any]]] = {
        "blocks": [
            {
                "type": "header",
                "text": {
                    "type": "plain_text",
                    "text": "THE FOLLOWING RISK HAS BEEN DETECTED BY THE DSPM :warning:",
                    "emoji": True,
                },
            },
            {
                "type": "section",
                "block_id": "section_incident_details",
                "text": {
                    "type": "mrkdwn",
                    "text": f"*XSOAR Incident ID:* {incident.get('incidentId')}\n"
                    f"*XSOAR Incident link:* {incidentLink}\n"
                    f"*DSPM Risk ID:* {incident.get('riskFindingId')}\n"
                    f"*Rule Name:* {incident.get('ruleName')}\n"
                    f"*Severity:* {incident.get('severity')}\n"
                    f"*Asset Name:* {incident.get('assetName')}\n"
                    f"*Asset ID:* {incident.get('assetId')}\n"
                    f"*Project ID:* {incident.get('projectId')}\n"
                    f"*Cloud Provider:* {incident.get('cloudProvider')}\n"
                    f"*Service Type:* {incident.get('serviceType')}\n"
                    f"*First Discovered:* {incident.get('firstDetectedOn')}\n"
                    f"*Remediate Instruction:* {incident.get('remediateInstruction')}\n",
                },
            },
            {"type": "divider"},
            {
                "type": "actions",
                "elements": [
                    {
                        "type": "radio_buttons",
                        "options": [
                            {
                                "text": {"type": "plain_text", "text": "Create a Jira ticket", "emoji": True},
                                "value": "Create a Jira ticket",
                            }
                        ],
                        "action_id": "actionId-0",
                    }
                ],
            },
            {
                "type": "input",
                "element": {
                    "type": "plain_text_input",
                    "placeholder": {"type": "plain_text", "text": "Please enter a valid Project Name", "emoji": True},
                    "action_id": "project_name",
                },
                "label": {"type": "plain_text", "text": "Enter Project Name", "emoji": True},
            },
            {
                "type": "input",
                "element": {
                    "type": "plain_text_input",
                    "placeholder": {"type": "plain_text", "text": "Please enter a valid Issue type", "emoji": True},
                    "action_id": "Issue_type",
                },
                "label": {"type": "plain_text", "text": "Enter Issue Type", "emoji": True},
            },
            {"type": "divider"},
        ]
    }

    # Log the block structure to verify correctness
    demisto.info(f"Block structure before modification: {json.dumps(block, indent=2)}")

    # Add the "Remediate a Risk" radio button option if ruleName exists in rule_names_dict
    if rule_name in rule_names_dict:
        try:
            # Ensure the 'actions' block exists at the correct index
            elements = block["blocks"][3].get("elements", [{}])[0].get("options", [])
            elements.insert(
                0,
                {
                    "text": {
                        "type": "plain_text",
                        "text": f"Remediate a Risk - {rule_names_dict.get(rule_name)}",
                        "emoji": True,
                    },
                    "value": "Remediate a Risk",
                },
            )
        except (AttributeError, IndexError, TypeError) as e:
            demisto.error(f"Error inserting 'Remediate a Risk' option: {str(e)}")
            raise
    res = {"block": block}
    return res


""" MAIN FUNCTION """


def main():  # pragma: no cover
    """
    The main function for creating and storing a Slack block message based on DSPM risk data.

    It processes the DSPM incident details, creates a Slack block and outputs the block.
    If the rule name matches certain conditions, a 'Remediate a Risk'
    option is added to the Slack block.

    Returns:
        None: Results are returned via demisto.results() and CommandResults().
    """
    rule_names_dict = {
        "Sensitive asset open to world": (
            "This action would block the public access to specific containers in storage account "
            "(Azure) or to the bucket (AWS/GCP) based on the cloud provider."
        ),
        "Empty storage asset": (
            "This action would delete the storage account (Azure) or the storage bucket (AWS/GCP) based on the cloud provider."
        ),
    }
    try:
        incident = demisto.args().get("dspmIncident")
        incidentLink = demisto.args().get("incidentLink")
        slackBlock = create_slack_block(incident, rule_names_dict, incidentLink)

        return_results(
            CommandResults(
                outputs_prefix="slackBlock",
                outputs=slackBlock,
            )
        )

    except Exception as excep:
        return_error(f"Failed to execute CreateDSPMRiskSlackBlocks. Error: {str(excep)}")


""" ENTRY POINT """

if __name__ in ("__main__", "__builtin__", "builtins"):  # pragma: no cover
    main()

README