# By: Riasat Ullah # This file contains variables and functions for assisting ServiceNow integration processes. from dbqueries.integrations import db_servicenow from requests.auth import HTTPBasicAuth from utils import constants, logging, var_names import json import requests # ServiceNow standard request headers standard_headers = {'Accept': 'application/json', 'Content-Type': 'application/json'} # url paths url_assignment_groups = '{0}/api/now/table/sys_user_group?sysparm_fields=name%2Csys_id' url_create_incident = '{0}/api/now/table/incident' url_edit_incident = '{0}/api/now/table/incident/{1}' url_view_incident = '{0}/task.do?sys_id={1}' # TaskCall allowed url calls cmd_acknowledge = 'acknowledge' cmd_add_conference_bridge = 'add-conference-bridge' cmd_add_responders = 'add-responders' cmd_create_incident = 'create-incident' cmd_list_conference_bridges = 'list-conference-bridges' cmd_list_policies = 'list-policies' cmd_list_workflows = 'list-workflows' cmd_notate = 'notate' cmd_reassign = 'reassign' cmd_resolve = 'resolve' cmd_run_workflow = 'run-workflow' cmd_unacknowledge = 'unacknowledge' cmd_update_status = 'update-status' # ServiceNow's status codes closed_state_code = '7' resolved_state_code = '6' # ServiceNow variables var_assignment_group = 'assignment_group' var_description = 'description' var_number = 'number' var_servicenow_incident_id = 'servicenow_incident_id' var_short_description = 'short_description' var_state = 'state' var_sys_id = 'sys_id' var_urgency = 'urgency' def servicenow_get_request(url, username, password, payload=None): ''' Makes a GET request to the api and returns the response. :param url: api url :param username: admin username :param password: admin password :param payload: (dict) query parameters :return: [status code, response] ''' try: response = requests.get(url, headers=standard_headers, auth=HTTPBasicAuth(username, password), params=payload) return [response.status_code, response.json()] except (ConnectionRefusedError, ConnectionError) as e: logging.exception('Failed to connect with ServiceNow') logging.exception(str(e)) raise ConnectionRefusedError except Exception as e: logging.exception('ServiceNow API request failed...') logging.exception(str(e)) raise ConnectionRefusedError def servicenow_post_request(url, username, password, body, status_only=False): ''' Makes a POST request to the api and returns the response. :param url: api url :param username: admin username :param password: admin password :param body: the JSON body to send with the request :param status_only: True if the json body of the response is not needed :return: [status code, response] ''' try: response = requests.post(url, headers=standard_headers, auth=HTTPBasicAuth(username, password), data=json.dumps(body)) if status_only: return response.status_code else: return [response.status_code, response.json()] except (ConnectionRefusedError, ConnectionError) as e: logging.exception('Failed to connect with ServiceNow') logging.exception(str(e)) raise ConnectionRefusedError except Exception as e: logging.exception('ServiceNow API request failed...') logging.exception(str(e)) raise ConnectionRefusedError def servicenow_patch_request(url, username, password, body): ''' Makes a PATCH request to the api and returns the response status. :param url: api url :param username: admin username :param password: admin password :param body: the JSON body to send with the request :return: [status code, response] ''' try: response = requests.patch(url, headers=standard_headers, auth=HTTPBasicAuth(username, password), data=json.dumps(body)) return response.status_code except (ConnectionRefusedError, ConnectionError) as e: logging.exception('Failed to connect with ServiceNow') logging.exception(str(e)) raise ConnectionRefusedError except Exception as e: logging.exception('ServiceNow API request failed...') logging.exception(str(e)) raise ConnectionRefusedError def get_taskcall_service_and_policy(integ_details, sn_grp): ''' Get the TaskCall service or escalation policy an assignment group is tied to. :param integ_details: additional info field of the integration :param sn_grp: ServiceNow assignment group :return: (tuple) -> service ID, escalation policy ID ''' tc_srv, tc_pol = None, None if integ_details[var_names.services] is not None and sn_grp in integ_details[var_names.services]: tc_srv = integ_details[var_names.services][sn_grp] elif integ_details[var_names.policies] is not None and sn_grp in integ_details[var_names.policies]: tc_pol = integ_details[var_names.policies][sn_grp] else: tc_srv = integ_details[var_names.default_service] return tc_srv, tc_pol def get_taskcall_urgency(integ_details, sn_urg): ''' Get the TaskCall urgency level from a ServiceNow urgency level. :param integ_details: additional info field of the integration :param sn_urg: ServiceNow urgency :return: (int) urgency level ''' urgency = constants.critical_urgency if var_names.urgency_level in integ_details and integ_details[var_names.urgency_level] is not None: for key in integ_details[var_names.urgency_level]: if integ_details[var_names.urgency_level][key][var_names.urgency_level] == sn_urg: urgency = int(key) break return urgency def create_servicenow_incident(conn, timestamp, org_id, integ_key, integ_info, inst_id, org_inst_id, task_title, text_msg, instance_state, urgency, notes): ''' Creates a ServiceNow incident. :param conn: db connection :param timestamp: timestamp when this request is being made :param org_id: organization ID :param integ_key: the integration key of this ServiceNow integration :param integ_info: additional info of this integration :param inst_id: instance ID (TaskCall) :param org_inst_id: organization instance ID (TaskCall) :param task_title: instance task title that will be used as the summary of the issue :param text_msg: instance text details that will be used as the description of the issue :param instance_state: the current state of the instance :param urgency: the urgency level of the instance :param notes: (list) notes added to the incident :return: (str) -> issue key (None if issue cannot be created) ''' srv_now_integ_det = db_servicenow.get_servicenow_instance_details(conn, timestamp, org_id, integration_key=integ_key) srv_now_host = srv_now_integ_det[var_names.vendor_endpoint] srv_now_admin = srv_now_integ_det[var_names.username] srv_now_pwd = srv_now_integ_det[var_names.password] srv_now_status_id = integ_info[var_names.status][instance_state] urg_det = integ_info[var_names.urgency_level][str(urgency)] srv_now_urgency_id = urg_det[var_names.urgency_level] srv_now_impact_id = urg_det[var_names.impact] body = { "x_116368_taskcall_taskcall_global_id": inst_id, "x_116368_taskcall_taskcall_incident_id": org_inst_id, "short_description": task_title, "description": text_msg, "urgency": srv_now_urgency_id, "impact": srv_now_impact_id, "state": srv_now_status_id, "work_notes": ' | '.join([x[var_names.notes] for x in notes]) if notes is not None else None, "caller_id": "abraham.lincoln@example.com" } status, output = servicenow_post_request(url_create_incident.format(srv_now_host), srv_now_admin, srv_now_pwd, body) srv_now_inc_id, srv_now_display_id = None, None if status in [200, 201]: srv_now_inc_id = output['result']['sys_id'] srv_now_display_id = output['result']['number'] else: logging.error('ServiceNow incident creation error (' + str(status) + ') - ' + str(output)) return srv_now_inc_id, status, {**{var_names.display_name: srv_now_display_id}, **output} def add_servicenow_work_note(conn, timestamp, org_id, integ_key, integ_info, srv_now_inc_id, comment): ''' Get the body of the request to add a comment to a ServiceNow incident. :param conn: db connection :param timestamp: timestamp when this request is being made :param org_id: organization ID :param integ_key: the integration key of this ServiceNow integration :param integ_info: additional info of this integration :param srv_now_inc_id: ServiceNow incident ID :param comment: (str) the comment ''' if integ_info[var_names.to_sync_notes]: srv_now_integ_det = db_servicenow.get_servicenow_instance_details(conn, timestamp, org_id, integration_key=integ_key) srv_now_host = srv_now_integ_det[var_names.vendor_endpoint] srv_now_admin = srv_now_integ_det[var_names.username] srv_now_pwd = srv_now_integ_det[var_names.password] body = {'work_notes': comment} servicenow_patch_request(url_edit_incident.format(srv_now_host, str(srv_now_inc_id)), srv_now_admin, srv_now_pwd, body) def update_servicenow_status(conn, timestamp, org_id, integ_key, integ_info, instance_state, srv_now_inc_id, user_email=None): ''' Update the status of a ServiceNow incident. :param conn: db connection :param timestamp: timestamp when this request is being made :param org_id: organization ID :param integ_key: the integration key of this ServiceNow integration :param integ_info: additional info of this integration :param instance_state: the current state of the instance :param srv_now_inc_id: ServiceNow incident ID :param user_email: email address of the user ''' srv_now_integ_det = db_servicenow.get_servicenow_instance_details(conn, timestamp, org_id, integration_key=integ_key) srv_now_host = srv_now_integ_det[var_names.vendor_endpoint] srv_now_admin = srv_now_integ_det[var_names.username] srv_now_pwd = srv_now_integ_det[var_names.password] srv_now_status_id = integ_info[var_names.status][instance_state] body = {'state': srv_now_status_id} if user_email is not None and instance_state == constants.acknowledged_state: body['assigned_to'] = user_email if srv_now_status_id in [resolved_state_code, closed_state_code]: body['resolved_by'] = user_email if user_email is not None else srv_now_admin body['close_code'] = 'Resolved by request' body['close_notes'] = 'Resolved by TaskCall' servicenow_patch_request(url_edit_incident.format(srv_now_host, str(srv_now_inc_id)), srv_now_admin, srv_now_pwd, body) def update_servicenow_urgency(conn, timestamp, org_id, integ_key, integ_info, urgency, srv_now_inc_id): ''' Update the status of a ServiceNow incident. :param conn: db connection :param timestamp: timestamp when this request is being made :param org_id: organization ID :param integ_key: the integration key of this ServiceNow integration :param integ_info: additional info of this integration :param urgency: TaskCall urgency :param srv_now_inc_id: ServiceNow incident ID ''' srv_now_integ_det = db_servicenow.get_servicenow_instance_details(conn, timestamp, org_id, integration_key=integ_key) srv_now_host = srv_now_integ_det[var_names.vendor_endpoint] srv_now_admin = srv_now_integ_det[var_names.username] srv_now_pwd = srv_now_integ_det[var_names.password] urg_det = integ_info[var_names.urgency_level][str(urgency)] srv_now_urgency_id = urg_det[var_names.urgency_level] srv_now_impact_id = urg_det[var_names.impact] body = {'urgency': srv_now_urgency_id, 'impact': srv_now_impact_id} servicenow_patch_request(url_edit_incident.format(srv_now_host, str(srv_now_inc_id)), srv_now_admin, srv_now_pwd, body)