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()