# By: Riasat Ullah # This file contains all constants and functions that are related to the integration with slack. from cache_queries import cache_organizations from dbqueries import db_services, db_users from dbqueries.integrations import db_slack from slack_sdk.errors import SlackApiError from threading import Thread from translators import label_translator as _lt from utils import constants, errors, helpers, info, integration_type_names, key_manager, label_names as lnm, \ logging, permissions, times, tokenizer, url_paths, var_names import datetime import json import requests # slack commands slack_command = '/taskcall' slack_create_incident_cmd = 'create-incident' slack_get_user_cmd = 'get-user' slack_help_cmd = 'help' slack_remove_user_cmd = 'remove-user' slack_verify_user_cmd = 'verify-user' # slack error messages slack_api_error = 'Slack could not process request. Please check if incoming-web-hook/response-url is still valid.' unknown_action_error = 'Unknown action received. Cannot process request.' unknown_command_error = 'Unknown command was received. Cannot process request' # slack modal action values open_modal_add_responders = 'open_modal_add_responders' open_modal_create_incident = 'open_modal_create_incident' open_modal_notate = 'open_modal_notate' open_modal_reassign = 'open_modal_reassign' open_modal_run_workflow = 'open_modal_run_workflow' open_modal_status_update = 'open_modal_status_update' # slack variables var_channel_id = 'slack_channel_id' var_channel_name = 'slack_channel_name' var_interactive_message = 'interactive_message' var_team_id = 'slack_team_id' var_team_name = 'slack_team_name' # slack url paths view_open_url = 'https://slack.com/api/views.open' view_update_url = 'https://slack.com/api/views.update' # default options for select tag default_select_options = [['All', 'All']] class SlackUserVerifier(Thread): def __init__(self, conn, slack_team_id, slack_channel_id, slack_user_id, taskcall_pref_name, slack_response_url): # check if slack team ID is associated to an organization or not # check if preferred username exists in the organization or not self.conn = conn self.slack_team_id = slack_team_id self.slack_channel_id = slack_channel_id self.slack_user_id = slack_user_id self.taskcall_pref_name = taskcall_pref_name self.slack_response_url = slack_response_url self.current_time = times.get_current_timestamp() Thread.__init__(self) def run(self): message = _lt.get_label(errors.err_verification_failed, constants.lang_en) try: org_id, service_id, integ_id, acc_token, slack_hook, external_info = db_slack.get_slack_integration_details( self.conn, self.current_time, self.slack_team_id, self.slack_channel_id) if service_id is not None: org_pref_maps = db_users.get_preferred_username_id_and_permissions(self.conn, self.current_time, org_id) if self.taskcall_pref_name in org_pref_maps: taskcall_user_id = org_pref_maps[self.taskcall_pref_name][0] if external_info is None: external_info = dict() if var_names.users not in external_info: external_info[var_names.users] = dict() external_info[var_names.users][self.slack_user_id] = taskcall_user_id db_services.update_organization_integration_type_details( self.conn, self.current_time, org_id, integration_type_names.slack, external_info, external_id=self.slack_team_id) message = _lt.get_label(info.msg_success, constants.lang_en) except KeyError as e: logging.exception(str(e)) except Exception as e: logging.exception(str(e)) message = _lt.get_label(errors.err_system_error, constants.lang_en) finally: data = {'text': message, 'replace_original': True} helpers.post_api_request(self.slack_response_url, data, get_status_only=True) class SlackUserRemover(Thread): ''' Removes an associated slack user from the database. ''' def __init__(self, conn, slack_team_id, slack_channel_id, slack_user_id, slack_response_url): self.conn = conn self.slack_team_id = slack_team_id self.slack_channel_id = slack_channel_id self.slack_user_id = slack_user_id self.slack_response_url = slack_response_url self.current_time = times.get_current_timestamp() Thread.__init__(self) def run(self): message = _lt.get_label(errors.err_unknown_resource, constants.lang_en) try: org_id, service_id, integ_id, acc_token, slack_hook, external_info = db_slack.get_slack_integration_details( self.conn, self.current_time, self.slack_team_id, self.slack_channel_id) if service_id is not None and external_info is not None and var_names.users in external_info\ and self.slack_user_id in external_info[var_names.users]: del external_info[var_names.users][self.slack_user_id] db_services.update_organization_integration_type_details( self.conn, self.current_time, org_id, integration_type_names.slack, external_info, external_id=self.slack_team_id) message = _lt.get_label(info.msg_success, constants.lang_en) except KeyError as e: logging.exception(str(e)) except Exception as e: logging.exception(str(e)) message = _lt.get_label(errors.err_system_error, constants.lang_en) finally: data = {'text': message, 'replace_original': True} helpers.post_api_request(self.slack_response_url, data, get_status_only=True) class SlackUserInfo(Thread): ''' Gets the TaskCall username of an associated Slack account. ''' def __init__(self, conn, slack_team_id, slack_channel_id, slack_user_id, slack_response_url): self.conn = conn self.slack_team_id = slack_team_id self.slack_channel_id = slack_channel_id self.slack_user_id = slack_user_id self.slack_response_url = slack_response_url self.current_time = times.get_current_timestamp() Thread.__init__(self) def run(self): message = _lt.get_label(errors.err_unknown_resource, constants.lang_en) try: org_id, service_id, integ_id, acc_token, slack_hook, external_info = db_slack.get_slack_integration_details( self.conn, self.current_time, self.slack_team_id, self.slack_channel_id) if service_id is not None and external_info is not None and var_names.users in external_info\ and self.slack_user_id in external_info[var_names.users]: req_user_id = external_info[var_names.users][self.slack_user_id] user_details = db_users.get_user_details(self.conn, self.current_time, org_id, user_id=req_user_id) if user_details is not None: message = user_details[var_names.preferred_username] except (KeyError, RuntimeError) as e: logging.exception(str(e)) except Exception as e: logging.exception(str(e)) message = _lt.get_label(errors.err_system_error, constants.lang_en) finally: data = {'text': message, 'replace_original': True} helpers.post_api_request(self.slack_response_url, data, get_status_only=True) class SlackIncidentDispatcher(object): ''' This is a slack incident details dispatcher class. This should be used to send incident details to a specific Slack channel. ''' def __init__(self, org_id, instance_id, org_instance_id, instance_title, text_msg, service_name, assigned_to_str, status, urgency, snapshots, slack_hook, slack_channel_id, acc_token, lang=constants.lang_en, cache=None, conference_bridge=None, slack_ts=None, user_display_name=None): self.org_id = org_id self.instance_id = instance_id self.org_instance_id = org_instance_id self.instance_title = instance_title self.text_msg = text_msg self.service_name = service_name self.assigned_to_str = assigned_to_str self.status = status self.urgency = urgency self.snapshots = snapshots self.slack_hook = slack_hook self.slack_channel_id = slack_channel_id self.acc_token = acc_token self.lang = lang self.conference_bridge = conference_bridge self.slack_ts = slack_ts self.user_display_name = user_display_name self.org_perm = None if cache is not None: self.org_perm = cache_organizations.get_single_organization_permission(cache, self.org_id) @staticmethod def make_button(label, value, action_id): button = { "type": "button", "text": { "type": "plain_text", "text": label, "emoji": True }, "value": value, "action_id": action_id } return button @staticmethod def make_overflow_option(label, value): option = { "text": { "type": "plain_text", "text": label, "emoji": True }, "value": value } return option def make_overflow_button(self): # Put together the actions that should be shown in the overflow button. actions = [ self.make_overflow_option(_lt.get_label(lnm.ttl_escalate, self.lang), constants.escalate_event), self.make_overflow_option(_lt.get_label(lnm.ttl_reassign, self.lang), open_modal_reassign) ] if self.org_perm is not None: if permissions.has_org_permission(self.org_perm, permissions.ORG_ADD_RESPONDERS_PERMISSION): actions.append(self.make_overflow_option(_lt.get_label(lnm.ttl_add_responders, self.lang), open_modal_add_responders)) if permissions.has_org_permission(self.org_perm, permissions.ORG_WORKFLOWS_PERMISSION): actions.append(self.make_overflow_option(_lt.get_label(lnm.ttl_run_workflow, self.lang), open_modal_run_workflow)) if permissions.has_org_permission(self.org_perm, permissions.ORG_INCIDENT_STATUS_PERMISSION): actions.append(self.make_overflow_option(_lt.get_label(lnm.ttl_new_update, self.lang), open_modal_status_update)) button = { "type": "overflow", "options": actions } return button def prepare_message(self): blocks = [ { "type": "section", "text": { "type": "mrkdwn", "text": "*<" + url_paths.web_incidents_details + '/' + str(self.org_instance_id) + "|#" + str(self.org_instance_id) + ": " + self.instance_title + ">*" } }, { "type": "section", "fields": [ { "type": "mrkdwn", "text": "*" + _lt.get_label(lnm.det_service, self.lang) + ":*\n" + self.service_name }, { "type": "mrkdwn", "text": "*" + _lt.get_label(lnm.det_urgency, self.lang) + ":*\n" + helpers.get_urgency_map(self.urgency, self.lang) } ] }, { "type": "section", "fields": [ { "type": "mrkdwn", "text": "*" + _lt.get_label(lnm.det_assigned_to, self.lang) + ":*\n" + self.assigned_to_str } ] } ] if self.conference_bridge is not None: fields = [] if var_names.conference_url in self.conference_bridge\ and self.conference_bridge[var_names.conference_url] is not None: fields.append({ "type": "mrkdwn", "text": "*" + _lt.get_label(lnm.ttl_conference_url, self.lang) + ":*\n" + "*<" + self.conference_bridge[var_names.conference_url] + "|" + self.conference_bridge[var_names.conference_url] + ">*" }) if var_names.conference_phone in self.conference_bridge\ and self.conference_bridge[var_names.conference_phone] is not None: fields.append({ "type": "mrkdwn", "text": "*" + _lt.get_label(lnm.ttl_dial_in_number, self.lang) + ":*\n" + "**" }) if len(fields) > 0: blocks.append({ "type": "section", "fields": fields }) if self.text_msg is not None and self.text_msg != '': blocks.append({ "type": "section", "text": { "type": "mrkdwn", "text": "*" + _lt.get_label(lnm.det_description, self.lang) + ":*\n" + self.text_msg } }) if self.snapshots is not None and len(self.snapshots) > 0: for img_url in self.snapshots: blocks.append({ "type": "image", "image_url": img_url, "alt_text": "snapshot" }) if self.status != constants.resolved_state: # Add actions to the list of blocks. blocks.append({ "type": "actions", "block_id": key_manager.encode_general_identifier( [self.acc_token, str(self.org_id), str(self.instance_id), str(self.org_instance_id)]), "elements": [ self.make_button(_lt.get_label(lnm.ttl_acknowledge, self.lang), constants.acknowledge_event, constants.acknowledge_event), self.make_button(_lt.get_label(lnm.ttl_resolve, self.lang), constants.resolve_event, constants.resolve_event), self.make_button(_lt.get_label(lnm.ttl_add_note, self.lang), open_modal_notate, open_modal_notate), self.make_overflow_button() ] }) border_color = '#DD0000' if self.status == constants.acknowledged_state: border_color = '#FCE205' elif self.status == constants.resolved_state: border_color = '#059142' data = { 'attachments': [ { 'color': border_color, 'blocks': blocks } ] } return data def basic_text(self): return "#" + str(self.org_instance_id) + ": " + self.instance_title + ' - ' + self.status def send_message(self): message_data = self.prepare_message() msg_ts, exe_status, exe_output = None, 400, None try: exe_status = send_by_webhook(self.slack_hook, message_data) except SlackApiError as e: logging.error('Error posting message to Slack: ' + str(e)) send_by_webhook(self.slack_hook, message_data) finally: return msg_ts, exe_status, exe_output class SlackIncidentActionProcessor(Thread): ''' Handles actions that are received from Slack interactive messages. ''' def __init__(self, conn, instance_id, org_instance_id, action, slack_team_id, slack_channel_id, slack_user_id, slack_username, slack_response_url, input_values, lang=constants.lang_en): self.conn = conn self.instance_id = instance_id self.org_instance_id = org_instance_id self.action = action self.slack_team_id = slack_team_id self.slack_channel_id = slack_channel_id self.slack_user_id = slack_user_id self.slack_username = slack_username self.slack_response_url = slack_response_url self.input_values = input_values self.lang = lang self.current_time = times.get_current_timestamp() Thread.__init__(self) def action_response_blocks(self, action_label): ''' Prepares the object that should be sent in response to an action. The response only describes what action was taken. :param action_label: label name of the string to convert to the user's language that describes the action :return: (dict) response body ''' inst_url_mrkdwn_text = "<" + url_paths.web_incidents_details + '/' + str(self.org_instance_id) + \ "|" + _lt.get_label(lnm.ttl_incident, self.lang) + \ "#" + str(self.org_instance_id) + ">: " message = [ { "type": "section", "text": { "type": "mrkdwn", "text": inst_url_mrkdwn_text + "*" + _lt.get_label(action_label, self.lang) + "* " +\ _lt.get_label(lnm.dsm_by, self.lang) + " " + self.slack_username } } ] return message def run(self): ''' Runs the action thread. Performs the actual action in TaskCall and sends a response. ''' data = { "replace_original": False, "response_type": "in_channel" } to_respond = True try: org_id, service_id, integ_id, acc_token, slack_hook, external_info =\ db_slack.get_slack_integration_details( self.conn, self.current_time, self.slack_team_id, self.slack_channel_id) if self.slack_response_url is None: self.slack_response_url = slack_hook if service_id is not None and external_info is not None and var_names.users in external_info\ and self.slack_user_id in external_info[var_names.users]: taskcall_user_id = external_info[var_names.users][self.slack_user_id] org_id, user_perm, org_perm = db_users.get_user_token_details( self.conn, self.current_time, taskcall_user_id)[:3] token_exp = times.get_current_timestamp() + datetime.timedelta(minutes=1) user_token = tokenizer.create_token(taskcall_user_id, org_id, user_perm, org_perm, token_exp) body = { var_names.instance_id: self.instance_id, var_names.access_method: constants.integrations_api } if self.action == constants.acknowledge_event: status, output = helpers.post_api_request(url_paths.incidents_acknowledge, body, user_token) if status == 200: to_respond = False else: data["text"] = output elif self.action == constants.resolve_event: status, output = helpers.post_api_request(url_paths.incidents_resolve, body, user_token) if status == 200: to_respond = False else: data["text"] = output elif self.action == constants.notate_event: body[var_names.notes] = self.input_values[var_names.body][var_names.data]['value'] status, output = helpers.post_api_request(url_paths.incidents_notate, body, user_token) if status == 200: to_respond = False else: data = {"text": output} elif self.action == constants.escalate_event: status, output = helpers.post_api_request(url_paths.incidents_escalate, body, user_token) if status == 200: data["blocks"] = self.action_response_blocks(lnm.evn_escalate) else: data = {"text": output} elif self.action == constants.reassign_event: body[var_names.reassign_to] = [ item['value'] for item in self.input_values[var_names.body][var_names.data]['selected_options'] ] status, output = helpers.post_api_request(url_paths.incidents_reassign, body, user_token) if status == 200: data = {"blocks": self.action_response_blocks(lnm.evn_reassign)} else: data = {"text": output} elif self.action == constants.add_responders_event: body[var_names.assignees] = [ item['value'] for item in self.input_values[var_names.body][var_names.data]['selected_options'] ] status, output = helpers.post_api_request(url_paths.incidents_add_responders, body, user_token) if status == 200: data = {"blocks": self.action_response_blocks(lnm.evn_add_responders)} else: data = {"text": output} elif self.action == constants.run_workflow_event: body[var_names.workflow_ref_id] = \ self.input_values[var_names.body][var_names.data]['selected_option']['value'] status, output = helpers.post_api_request(url_paths.incidents_run_workflow, body, user_token) if status == 200: data = {"blocks": self.action_response_blocks(lnm.evn_run_workflow)} else: data = {"text": output} elif self.action == constants.status_update_event: body[var_names.status_update] = self.input_values[var_names.body][var_names.data]['value'] status, output = helpers.post_api_request(url_paths.incidents_update_status, body, user_token) if status == 200: data = {"blocks": self.action_response_blocks(lnm.evn_status_update)} else: data = {"text": output} else: data["text"] = unknown_action_error else: data['text'] = _lt.get_label(errors.err_verification_failed, constants.lang_en) except KeyError as e: logging.exception(str(e)) data["text"] = _lt.get_label(str(e), constants.lang_en) except Exception as e: logging.exception(str(e)) data["text"] = _lt.get_label(errors.err_system_error, constants.lang_en) finally: if to_respond: status, output = helpers.post_api_request(self.slack_response_url, data) if status != 200 or (status == 200 and not output['ok']): logging.error(slack_api_error) logging.error(output) def send_by_webhook(slack_hook, message): try: header_params = {'Accept': 'application/json', 'Content-Type': 'application/json'} resp = requests.post(slack_hook, headers=header_params, data=json.dumps(message)) if resp.status_code != 200: logging.error(resp.text) return resp.status_code except Exception as e: logging.exception(str(e)) raise def send_event_update(slack_hook, evn_type, org_inst_id, inst_title, lang=constants.lang_en, user_display_name=None, extra_info=None): ''' Notify a slack channel about an action that has been performed on an incident. :param slack_hook: the Slack endpoint to send the message to :param evn_type: type of event (ACKNOWLEDGE, RESOLVE, etc) :param org_inst_id: organization instance ID :param inst_title: title of the instance :param lang: language the message should be sent in :param user_display_name: name of the user who is taking the action :param extra_info: extra details to add in the chat message :return: status of the HTTP request ''' if evn_type == constants.acknowledge_event: evn_lbl = lnm.evn_acknowledge elif evn_type == constants.notate_event: evn_lbl = lnm.evn_notate elif evn_type == constants.resolve_event: evn_lbl = lnm.evn_resolve elif evn_type == constants.status_update_event: evn_lbl = lnm.evn_status_update elif evn_type == constants.un_acknowledge_event: evn_lbl = lnm.evn_un_acknowledge elif evn_type == constants.urgency_amendment_event: evn_lbl = lnm.evn_urgency_amendment else: return message_data = {'blocks': get_action_response_blocks(org_inst_id, evn_lbl, user_display_name, inst_title, lang, extra_info)} exe_status = send_by_webhook(slack_hook, message_data) return exe_status def get_action_response_blocks(org_inst_id, action_label, event_by_name, inst_title=None, lang=constants.lang_en, extra_info=None): ''' Prepares the object that should be sent in response to an action. The response only describes what action was taken. :param org_inst_id: organization instance ID :param action_label: label name of the string to convert to the user's language that describes the action :param event_by_name: name of person who performed the action :param inst_title: title of the incident :param lang: language the content should be sent in :param extra_info: extra details to add in the chat message :return: (dict) response body ''' trimmed_title = '' if inst_title is not None: trimmed_title = inst_title[:40] + '...' if len(inst_title) > 40 else inst_title inst_url_mrkdwn_text = "<" + url_paths.web_incidents_details + '/' + str(org_inst_id) + \ "|" + _lt.get_label(lnm.ttl_incident, lang) + " #" + str(org_inst_id) + "> " + \ trimmed_title + ": " action_by_str = '' if event_by_name is not None: action_by_str += _lt.get_label(lnm.dsm_by, lang) + " " + event_by_name if extra_info is not None: action_by_str += ' - ' + extra_info message = [ { "type": "section", "text": { "type": "mrkdwn", "text": inst_url_mrkdwn_text + "*" + _lt.get_label(action_label, lang) + "* " + action_by_str } } ] return message def get_create_incident_button_content(org_id, acc_token, lang=constants.lang_en): ''' Prepares the content needed to display a create incident button. This is sent after a user submits the create-incident command. :param org_id: ID of the organization :param acc_token: Slack access token for the integration :param lang: language the content should be in :return: (dict) message content ''' return { "blocks": [ { "type": "section", "text": { "type": "plain_text", "text": _lt.get_label(info.msg_slack_create_incident, lang) } }, { "type": "actions", "block_id": key_manager.encode_general_identifier([acc_token, str(org_id)]), "elements": [ { "type": "button", "text": { "type": "plain_text", "text": _lt.get_label(lnm.ttl_create_incident, lang), "emoji": True }, "style": "primary", "value": open_modal_create_incident, "action_id": open_modal_create_incident } ] } ] } def textbox_modal_view(event_type, metadata, modal_title, submit_btn_label, lang=constants.lang_en): ''' Gets the Slack view object for a modal that only has a textbox. :param event_type: the type of event that will be triggered by this modal :param metadata: metadata to pass to the view :param modal_title: the title of the modal :param submit_btn_label: the label of the submit button :param lang: language the modal labels should be in :return: (dict) Slack view object ''' return { "view": { "type": "modal", "callback_id": event_type, "private_metadata": metadata, "title": { "type": "plain_text", "text": modal_title, "emoji": True }, "blocks": [ { "type": "input", "block_id": var_names.body, "label": { "type": "plain_text", "text": modal_title, "emoji": True }, "element": { "type": "plain_text_input", "multiline": True, "action_id": var_names.data } } ], "submit": { "type": "plain_text", "text": submit_btn_label, "emoji": True }, "close": { "type": "plain_text", "text": _lt.get_label(lnm.ins_cancel, lang), "emoji": True } } } def select_modal_view(event_type, metadata, modal_title, submit_btn_label, data, is_multi=True, lang=constants.lang_en): ''' Gets the Slack view object for a modal with a single select tag. :param event_type: the type of event that will be triggered by this modal :param metadata: metadata to pass to the view :param modal_title: the title of the modal :param submit_btn_label: the label of the submit button :param data: data to populate the select tag with :param is_multi: (boolean) if multiple selections will be allowed or not :param lang: language the modal labels should be in :return: (dict) Slack view object ''' if len(data) == 0: data = default_select_options return { "view": { "type": "modal", "callback_id": event_type, "private_metadata": metadata, "title": { "type": "plain_text", "text": modal_title, "emoji": True }, "blocks": [ { "type": "input", "block_id": var_names.body, "label": { "type": "plain_text", "text": modal_title, "emoji": True }, "element": { "type": "multi_static_select" if is_multi else "static_select", "action_id": var_names.data, "placeholder": { "type": "plain_text", "text": _lt.get_label(lnm.ins_select, lang), "emoji": True }, "options": [ { "text": { "type": "plain_text", "text": item[0], "emoji": True }, "value": item[1] } for item in data ] } } ], "submit": { "type": "plain_text", "text": submit_btn_label, "emoji": True }, "close": { "type": "plain_text", "text": _lt.get_label(lnm.ins_cancel, lang), "emoji": True } } } def create_incident_modal_content(metadata, services_basic_list, lang=constants.lang_en): ''' Create the modal view for creating incidents. :param metadata: metadata to pass to the view :param services_basic_list: list of services :param lang: language the modal labels should be in :return: (dict) Slack view object ''' if len(services_basic_list) == 0: services_basic_list = default_select_options return { "view": { "type": "modal", "callback_id": constants.trigger_event, "private_metadata": metadata, "title": { "type": "plain_text", "text": _lt.get_label(lnm.ttl_create_incident, lang), "emoji": True }, "blocks": [ { "type": "input", "block_id": var_names.title, "label": { "type": "plain_text", "text": _lt.get_label(lnm.ttl_incident_title, lang), "emoji": True }, "element": { "type": "plain_text_input", "action_id": var_names.title } }, { "type": "input", "block_id": var_names.service, "label": { "type": "plain_text", "text": _lt.get_label(lnm.det_service, lang), "emoji": True }, "element": { "type": "static_select", "action_id": var_names.service, "placeholder": { "type": "plain_text", "text": _lt.get_label(lnm.ins_select, lang) + "...", "emoji": True }, "options": [ { "text": { "type": "plain_text", "text": item[0], "emoji": True }, "value": item[1] } for item in services_basic_list ] } }, { "type": "input", "block_id": var_names.urgency_level, "label": { "type": "plain_text", "text": _lt.get_label(lnm.det_urgency, lang), "emoji": True }, "element": { "type": "static_select", "action_id": var_names.urgency_level, "placeholder": { "type": "plain_text", "text": _lt.get_label(lnm.ins_select, lang) + "...", "emoji": True }, "options": [ { "text": { "type": "plain_text", "text": item[0], "emoji": True }, "value": item[1] } for item in [ [_lt.get_label(lnm.opt_minor, lang), '1'], [_lt.get_label(lnm.opt_low, lang), '2'], [_lt.get_label(lnm.opt_medium, lang), '3'], [_lt.get_label(lnm.opt_high, lang), '4'], [_lt.get_label(lnm.opt_critical, lang), '5'] ] ] } }, { "type": "input", "block_id": var_names.text_msg, "label": { "type": "plain_text", "text": _lt.get_label(lnm.det_description, lang), "emoji": True }, "element": { "type": "plain_text_input", "multiline": True, "action_id": var_names.text_msg } } ], "submit": { "type": "plain_text", "text": _lt.get_label(lnm.ttl_create_incident, lang), "emoji": True }, "close": { "type": "plain_text", "text": _lt.get_label(lnm.ins_cancel, lang), "emoji": True } } } def open_modal_view(acc_token, trigger_id, data): ''' Open modal view in Slack. :param acc_token: access token needed for the call :param trigger_id: trigger ID received from Slack :param data: the modal body :return: ID of the view that was created ''' header_params = { 'Accept': 'application/json', 'Content-Type': 'application/json; charset=utf-8', 'Authorization': 'Bearer ' + acc_token } data['trigger_id'] = trigger_id response = requests.post(view_open_url, headers=header_params, data=json.dumps(data)) status, output = response.status_code, response.json() if status == 200 and output['ok']: return output['view']['id'], output['view']['hash'] else: logging.error('Failed to open Slack modal...') logging.error(output) return None, None def update_modal_view(acc_token, view_id, hash_val, data): ''' Update modal view in Slack. :param acc_token: access token needed for the call :param view_id: ID of the view; this will be needed to update the view :param hash_val: the value of the hash of the original view; will be needed to update the view :param data: the modal body ''' header_params = { 'Accept': 'application/json', 'Content-Type': 'application/json; charset=utf-8', 'Authorization': 'Bearer ' + acc_token } data['view_id'] = view_id data['hash'] = hash_val response = requests.post(view_update_url, headers=header_params, data=json.dumps(data)) status, output = response.status_code, response.json() if status != 200 or (status == 200 and not output['ok']): logging.error('Failed to update Slack modal...') logging.error(output) def send_slack_message(slack_hook, message): ''' Send a message to a Slack channel. :param slack_hook: the Slack endpoint to send the message to. :param message: message to send :return: ''' data = { 'blocks': [ { "type": "section", "text": { "type": "mrkdwn", "text": message } } ] } try: send_by_webhook(slack_hook, data) except Exception as e: logging.exception(str(e)) raise