# By: Riasat Ullah # This file contains functions for handling db queries for incident workflows. from dbqueries import db_business_services, db_integrations, db_policies, db_services, db_teams, db_users from exceptions.user_exceptions import DependencyFound from psycopg2 import errorcodes from utils import constants, errors, key_manager, permissions, var_names from validations import string_validator import configuration as configs import copy import datetime import json import psycopg2 def create_workflow(conn, timestamp, organization_id, workflow_name, description, is_manual, auto_triggers, workflow_steps, with_run_team, with_edit_team): ''' Create an incident workflow. :param conn: db connection :param timestamp: timestamp when this request is being made :param organization_id: ID of the organization this workflow is for :param workflow_name: name to identify the workflow with :param description: description of what the workflow is for :param is_manual: (boolean) True if the workflow is allowed to be run manually :param auto_triggers: (list) if the workflow should be run automatically then the list of events after which it should be triggered :param workflow_steps: (dict) actions that should be taken on the workflow :param with_run_team: ID of the team that can run the workflow if the workflow has only team specific run permissions :param with_edit_team: ID of the team that can run the workflow if the workflow has only team specific edit permissions :errors: AssertionError, DatabaseError, KeyError, ValueError ''' assert isinstance(timestamp, datetime.datetime) assert isinstance(organization_id, int) run_tm_id, edit_tm_id, workflow_steps = validate_workflow( conn, timestamp, organization_id, workflow_name, description, is_manual, auto_triggers, workflow_steps, with_run_team, with_edit_team ) query = ''' select create_workflow( %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s ); ''' query_params = (timestamp, constants.end_timestamp, organization_id, key_manager.generate_reference_key(), workflow_name, description, is_manual, auto_triggers, json.dumps(workflow_steps), run_tm_id, edit_tm_id,) try: conn.execute(query, query_params) except psycopg2.DatabaseError: raise def edit_workflow(conn, timestamp, organization_id, workflow_ref_id, workflow_name, description, is_manual, auto_triggers, workflow_steps, with_run_team, with_edit_team): ''' Edits an existing workflow. :param conn: db connection :param timestamp: timestamp when this request is being made :param organization_id: ID of the organization this workflow is for :param workflow_ref_id: reference ID of the workflow that is being edited :param workflow_name: name to identify the workflow with :param description: description of what the workflow is for :param is_manual: (boolean) True if the workflow is allowed to be run manually :param auto_triggers: (list) if the workflow should be run automatically then the list of events after which it should be triggered :param workflow_steps: (dict) actions that should be taken on the workflow :param with_run_team: ID of the team that can run the workflow if the workflow has only team specific run permissions :param with_edit_team: ID of the team that can run the workflow if the workflow has only team specific edit permissions :errors: AssertionError, DatabaseError, KeyError, ValueError ''' assert isinstance(timestamp, datetime.datetime) assert isinstance(organization_id, int) unmasked_workflow_ref_id = key_manager.unmask_reference_key(workflow_ref_id) run_tm_id, edit_tm_id, workflow_steps = validate_workflow( conn, timestamp, organization_id, workflow_name, description, is_manual, auto_triggers, workflow_steps, with_run_team, with_edit_team ) query = ''' select edit_workflow ( %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s ); ''' query_params = (timestamp, constants.end_timestamp, organization_id, unmasked_workflow_ref_id, workflow_name, description, is_manual, auto_triggers, json.dumps(workflow_steps), run_tm_id, edit_tm_id,) try: conn.execute(query, query_params) except psycopg2.IntegrityError as e: if e.pgcode == errorcodes.CHECK_VIOLATION: raise LookupError(errors.err_unknown_resource) except psycopg2.DatabaseError: raise def delete_workflow(conn, timestamp, organization_id, workflow_ref_id): ''' Delete a workflow. :param conn: db connection :param timestamp: timestamp when this request is being made :param organization_id: ID of the organization the workflow belongs to :param workflow_ref_id: reference ID of the workflow :errors: AssertionError, DatabaseError, LookupError ''' assert isinstance(timestamp, datetime.datetime) assert isinstance(organization_id, int) unmasked_workflow_ref_id = key_manager.unmask_reference_key(workflow_ref_id) query = ''' begin; update workflows set end_timestamp = %(timestamp)s where start_timestamp <= %(timestamp)s and end_timestamp > %(timestamp)s and organization_id = %(org_id)s and workflow_ref_id = %(ref_id)s; end; ''' query_params = {'timestamp': timestamp, 'org_id': organization_id, 'ref_id': unmasked_workflow_ref_id} try: conn.execute(query, query_params) except psycopg2.IntegrityError as e: if e.pgcode == errorcodes.CHECK_VIOLATION: raise LookupError(errors.err_unknown_resource) except psycopg2.DatabaseError: raise def enable_workflow(conn, timestamp, organization_id, workflow_ref_id, to_enable): ''' Enables or disables a workflow. :param conn: db connection :param timestamp: timestamp when this request is being made :param organization_id: ID of the organization the workflow belongs to :param workflow_ref_id: reference ID of the workflow :param to_enable: True if the workflow should be enabled; False otherwise :errors: AssertionError, DatabaseError, LookupError, PermissionError, ValueError ''' assert isinstance(organization_id, int) assert isinstance(to_enable, bool) unmasked_workflow_ref_id = key_manager.unmask_reference_key(workflow_ref_id) query = "select enable_workflow(%s, %s, %s, %s, %s);" query_params = (timestamp, constants.end_timestamp, organization_id, unmasked_workflow_ref_id, to_enable,) try: conn.execute(query, query_params) except psycopg2.IntegrityError as e: if e.pgcode == errorcodes.CHECK_VIOLATION: raise LookupError(errors.err_unknown_resource) elif e.pgcode == errorcodes.RESTRICT_VIOLATION: raise PermissionError(errors.err_user_rights) else: raise except psycopg2.DatabaseError: raise def has_run_permission(conn, timestamp, organization_id, workflow_id, user_id, is_global_responder): ''' Check if a user has permissions to run a workflow. :param conn: db connection :param timestamp: timestamp when this request is being made :param organization_id: ID of the organization the workflow belongs to :param workflow_id: ID of the workflow :param user_id: ID of the user requesting the edit :param is_global_responder: True if the user is has responder rights; False otherwise :errors: AssertionError, DatabaseError, LookupError, PermissionError, ValueError ''' assert isinstance(organization_id, int) assert isinstance(workflow_id, int) assert isinstance(user_id, int) query = "select has_workflow_permission(%s, %s, %s, %s, %s, false, %s);" query_params = (timestamp, organization_id, workflow_id, user_id, permissions.COMPONENT_ADVANCED_RESPOND_PERMISSION, is_global_responder,) try: result = conn.fetch(query, query_params) return result[0][0] except psycopg2.IntegrityError as e: if e.pgcode == errorcodes.CHECK_VIOLATION: raise LookupError(errors.err_unknown_resource) elif e.pgcode == errorcodes.RESTRICT_VIOLATION: raise PermissionError(errors.err_user_rights) else: raise except psycopg2.DatabaseError: raise def has_edit_permission(conn, timestamp, organization_id, workflow_id, user_id, is_global_admin): ''' Check if a user has permissions to run a workflow. :param conn: db connection :param timestamp: timestamp when this request is being made :param organization_id: ID of the organization the workflow belongs to :param workflow_id: ID of the workflow :param user_id: ID of the user requesting the edit :param is_global_admin: True if the user is an organization admin; False otherwise :errors: AssertionError, DatabaseError, LookupError, PermissionError, ValueError ''' assert isinstance(organization_id, int) assert isinstance(workflow_id, int) assert isinstance(user_id, int) query = "select has_workflow_permission(%s, %s, %s, %s, %s, %s, false);" query_params = (timestamp, organization_id, workflow_id, user_id, permissions.COMPONENT_ADVANCED_EDIT_PERMISSION, is_global_admin,) try: result = conn.fetch(query, query_params) return result[0][0] except psycopg2.IntegrityError as e: if e.pgcode == errorcodes.CHECK_VIOLATION: raise LookupError(errors.err_unknown_resource) elif e.pgcode == errorcodes.RESTRICT_VIOLATION: raise PermissionError(errors.err_user_rights) else: raise except psycopg2.DatabaseError: raise def get_workflow_id_from_ref_id(conn, timestamp, organization_id, workflow_ref_id): ''' Get the workflow ID given a reference ID. :param conn: db connection :param timestamp: timestamp when this request is being made :param organization_id: ID of the organization :param workflow_ref_id: reference ID of the workflow :return: (int) workflow ID ''' assert isinstance(timestamp, datetime.datetime) assert isinstance(organization_id, int) unmasked_ref_id = key_manager.unmask_reference_key(workflow_ref_id) query = ''' select workflow_id from workflows where start_timestamp <= %(timestamp)s and end_timestamp > %(timestamp)s and organization_id = %(org_id)s and workflow_ref_id = %(ref_id)s; ''' query_params = {'timestamp': timestamp, 'org_id': organization_id, 'ref_id': unmasked_ref_id} try: result = conn.fetch(query, query_params) if len(result) > 0: return result[0][0] else: raise LookupError(errors.err_unknown_resource) except psycopg2.DatabaseError: raise def get_workflows(conn, timestamp, organization_id, workflow_ids=None, workflow_ref_id=None, with_ref_id_maps=True, is_enabled=None): ''' Get the workflows of an organization. :param conn: db connection :param timestamp: timestamp when this request is being made :param organization_id: ID of the organization :param workflow_ids: (list) of workflow IDs (internal IDs) :param workflow_ref_id: reference ID of the workflow :param with_ref_id_maps: True if actionable items should be retrieved as their display and reference IDs :param is_enabled: None -> get all the workflows; True -> only enabled ones; False -> non-enabled ones :return: (list of dict) -> [ { workflow details }, ... ] ''' assert isinstance(timestamp, datetime.datetime) assert isinstance(organization_id, int) query_params = {'timestamp': timestamp, 'org_id': organization_id} flow_cond = [] if workflow_ids is not None: assert isinstance(workflow_ids, int) flow_cond.append('workflow_id = any(%(flow_id)s)') query_params['flow_id'] = workflow_ids if workflow_ref_id is not None: flow_cond.append('workflow_ref_id = %(ref_id)s') query_params['ref_id'] = key_manager.unmask_reference_key(workflow_ref_id) if is_enabled is not None: assert isinstance(is_enabled, bool) flow_cond.append("is_enabled = %(is_enb)s") query_params['is_enb'] = is_enabled query = ''' select workflow_id, workflow_ref_id, workflow_name, description, is_manual, trigger_methods, workflow_steps, run_permission_team_id, edit_permission_team_id, is_enabled from workflows where start_timestamp <= %(timestamp)s and end_timestamp > %(timestamp)s and organization_id = %(org_id)s {0}; '''.format(' and ' + ' and '.join(flow_cond) if len(flow_cond) > 0 else '') try: result = conn.fetch(query, query_params) data = [] for id_, ref_, name_, desc_, manual_, trig_meth_, steps_, run_perm_, edit_perm_, is_enb_ in result: flow_details = { var_names.workflow_id: id_, var_names.workflow_ref_id: key_manager.conceal_reference_key(ref_) if with_ref_id_maps else ref_, var_names.workflow_name: name_, var_names.description: desc_, var_names.is_manual: manual_, var_names.trigger_method: trig_meth_, var_names.run_permissions: run_perm_, var_names.edit_permissions: edit_perm_, var_names.is_enabled: is_enb_ } if with_ref_id_maps: flow_details[var_names.workflow] = externalize_workflow(conn, timestamp, organization_id, steps_) else: flow_details[var_names.workflow] = steps_ data.append(flow_details) return data except psycopg2.DatabaseError: raise def get_workflows_list(conn, timestamp, organization_id): ''' Get the workflows of an organization. :param conn: db connection :param timestamp: timestamp when this request is being made :param organization_id: ID of the organization :return: (list of dict) -> [ { workflow details }, ... ] ''' assert isinstance(timestamp, datetime.datetime) assert isinstance(organization_id, int) query = ''' with t1 as ( select workflow_id, workflow_ref_id, workflow_name, description, is_manual, trigger_methods, workflow_steps, run_permission_team_id, edit_permission_team_id, is_enabled, org.organization_timezone from workflows as wfl join organizations as org on org.organization_id = wfl.organization_id and org.start_timestamp <= %(timestamp)s and org.end_timestamp > %(timestamp)s where wfl.start_timestamp <= %(timestamp)s and wfl.end_timestamp > %(timestamp)s and wfl.organization_id = %(org_id)s ) , t2 as ( select team_id, team_ref_id, team_name from teams where start_timestamp <= %(timestamp)s and end_timestamp > %(timestamp)s and organization_id = %(org_id)s ) , t3 as ( select t1.*, t2.team_ref_id as run_team_ref, t2.team_name as run_team_name from t1 left join t2 on t1.run_permission_team_id is not null and t1.run_permission_team_id = t2.team_id ) , t4 as ( select t3.*, t2.team_ref_id as edit_team_ref, t2.team_name as edit_team_name from t3 left join t2 on t3.edit_permission_team_id is not null and t3.edit_permission_team_id = t2.team_id ) , t5 as ( select workflow_id, max(action_timestamp) as last_run from instance_workflows where workflow_id in (select workflow_id from t4) and action_timestamp <= %(timestamp)s group by workflow_id ) select workflow_ref_id, workflow_name, description, is_manual, trigger_methods, workflow_steps, is_enabled, run_team_ref, run_team_name, edit_team_ref, edit_team_name, case when t5.last_run is not null then (t5.last_run at time zone 'UTC' at time zone t4.organization_timezone)::timestamp else t5.last_run end as last_run from t4 left join t5 using(workflow_id); ''' query_params = {'timestamp': timestamp, 'org_id': organization_id} try: result = conn.fetch(query, query_params) data = [] for ref_, name_, desc_, manual_, trig_meth_, steps_, is_enb_, run_tm_ref_, run_tm_name_,\ edit_tm_ref_, edit_tm_name_, last_run_ in result: run_tm = [run_tm_name_, key_manager.conceal_reference_key(run_tm_ref_)]\ if run_tm_ref_ is not None and run_tm_name_ is not None else None edit_tm = [edit_tm_name_, key_manager.conceal_reference_key(edit_tm_ref_)]\ if edit_tm_ref_ is not None and edit_tm_name_ is not None else None ext_steps_ = externalize_workflow(conn, timestamp, organization_id, steps_) is_cond, act_list = False, [] for item in ext_steps_: is_cond, act_list = list_workflow_step_conditions_and_actions(item, is_cond, act_list) flow_details = { var_names.workflow_ref_id: key_manager.conceal_reference_key(ref_), var_names.workflow_name: name_, var_names.description: desc_, var_names.is_manual: manual_, var_names.trigger_method: trig_meth_, var_names.run_permissions: run_tm, var_names.edit_permissions: edit_tm, var_names.is_enabled: is_enb_, var_names.last_run: last_run_, var_names.conditions: is_cond, var_names.actions: list(set(act_list)) } data.append(flow_details) return data except psycopg2.DatabaseError: raise def get_basic_workflows_list(conn, timestamp, organization_id): ''' Get the basic list of workflows. :param conn: db connection :param timestamp: timestamp this request is being made on :param organization_id: ID of the organization :return: (list of list) -> [ [workflow name, workflow ref id], ... ] :errors: AssertionError, DatabaseError ''' assert isinstance(timestamp, datetime.datetime) assert isinstance(organization_id, int) query = ''' select workflow_name, workflow_ref_id from workflows where start_timestamp <= %s and end_timestamp > %s and organization_id = %s and is_enabled = true; ''' query_params = (timestamp, timestamp, organization_id,) try: result = conn.fetch(query, query_params) data = [] for name_, key_ in result: data.append([name_, key_manager.conceal_reference_key(key_)]) return data except psycopg2.DatabaseError: raise def get_workflow_last_run_time(conn, timestamp, organization_id, workflow_id): ''' Get the last time a workflow was run. :param conn: db connection :param timestamp: timestamp when this request is being made :param organization_id: ID of the organization the workflow belongs to :param workflow_id: ID of the workflow :return: (timestamp) last run timestamp :errors: AssertionError, DatabaseError ''' assert isinstance(timestamp, datetime.datetime) assert isinstance(organization_id, int) assert isinstance(workflow_id, int) query = ''' with t1 as ( select organization_id, organization_timezone from organizations where organization_id = %(org_id)s and start_timestamp <= %(timestamp)s and end_timestamp > %(timestamp)s ) , t2 as ( select wfl.organization_id, wfl.workflow_id, max(action_timestamp) as last_run from workflows as wfl join instance_workflows as inl on inl.workflow_id = wfl.workflow_id and inl.action_timestamp <= %(timestamp)s where wfl.workflow_id = %(flow_id)s and wfl.start_timestamp <= %(timestamp)s and wfl.end_timestamp > %(timestamp)s group by wfl.organization_id, wfl.workflow_id ) select t2.workflow_id, case when t2.last_run is not null then (t2.last_run at time zone 'UTC' at time zone t1.organization_timezone)::timestamp else t2.last_run end as last_run from t2 join t1 using (organization_id); ''' query_params = {'org_id': organization_id, 'flow_id': workflow_id, 'timestamp': timestamp} try: result = conn.fetch(query, query_params) if len(result) == 0: return None else: return result[0][1] except psycopg2.DatabaseError: raise def validate_workflow(conn, timestamp, org_id, workflow_name, description, is_manual, auto_triggers, workflow_steps, with_run_team, with_edit_team): ''' Validates the entire workflow and returns the internalized workflow steps. It uses other functions to validate each part of the workflow. :param conn: db connection :param timestamp: timestamp when this request is being made :param org_id: ID of the organization to the workflow belongs to :param workflow_name: name of the workflow :param description: description of the workflow :param is_manual: (boolean) True if the workflow should be allowed to be run manually; False otherwise :param auto_triggers: (list) of events after which the workflow should be run automatically :param workflow_steps: (list) of workflow steps :param with_run_team: (integer) if the workflow has team specific run permissions, then the ID of the team :param with_edit_team: (integer) if the workflow has team specific edit permissions, then the ID of the team :return: (list) of internalized workflow steps :error: AssertionError, DatabaseError, KeyError, ValueError ''' assert string_validator.is_standard_name(workflow_name) if description is not None: assert isinstance(description, str) assert isinstance(is_manual, bool) if auto_triggers is not None: assert isinstance(auto_triggers, list) assert set(auto_triggers).issubset({ constants.trigger_event, constants.acknowledge_event, constants.notate_event, constants.resolve_event, constants.urgency_amendment_event }) srv_dict = db_services.list_service_ids_from_ref_ids(conn, timestamp, org_id, as_dict=True) pol_dict = db_policies.list_policy_ids_from_ref_ids(conn, timestamp, org_id, as_dict=True) user_dict = db_users.get_user_ids_from_preferred_usernames(conn, timestamp, org_id, as_dict=True) team_dict = db_teams.get_team_ids_from_ref_ids(conn, timestamp, org_id, as_dict=True) bus_srv_dict = db_business_services.get_business_service_ids_from_ref_ids(conn, timestamp, org_id) cust_integ_dict = db_integrations.list_custom_action_integrations_from_keys(conn, timestamp, org_id, as_dict=True) chat_integ_dict = db_integrations.list_chat_integrations_from_keys(conn, timestamp, org_id, as_dict=True) run_team_id, edit_team_id = None, None if with_run_team is not None: unm_run = key_manager.unmask_reference_key(with_run_team) assert unm_run in team_dict run_team_id = team_dict[unm_run] if with_edit_team is not None: unm_edit = key_manager.unmask_reference_key(with_edit_team) assert unm_edit in team_dict edit_team_id = team_dict[unm_edit] new_steps = [] for step in workflow_steps: new_steps.append(validate_and_internalize_workflow_step( step, srv_dict, pol_dict, user_dict, team_dict, bus_srv_dict, cust_integ_dict, chat_integ_dict)) return run_team_id, edit_team_id, new_steps def validate_and_internalize_workflow_step(step, srv_dict, pol_dict, user_dict, team_dict, bus_srv_dict, cust_integ_dict, chat_integ_dict): ''' Validates a single workflow step and internalizes the values from reference IDs to IDs. :param step: (dict) single workflow step :param srv_dict: (dict) -> {ref id: service ID, ...} :param pol_dict: (dict) -> {ref id: policy ID, ...} :param user_dict: (dict) -> {pref name: user ID, ...} :param team_dict: (dict) -> {ref id: team ID, ...} :param bus_srv_dict: (dict) -> {ref ID: business service ID, ...} :param cust_integ_dict: (dict) -> {integration key: integration ID, ...} :param chat_integ_dict: (dict) -> {integration key: integration ID, ...} :return: (dict) -> internalized workflow step :error: AssertionError, DatabaseError, KeyError, ValueError ''' if var_names.actions in step: step = validate_and_internalize_workflow_action(step, pol_dict, user_dict, team_dict, bus_srv_dict, cust_integ_dict, chat_integ_dict) else: if var_names.block_if in step: assert var_names.conditions in step[var_names.block_if] assert var_names.workflow in step[var_names.block_if] # handle conditions step[var_names.block_if][var_names.conditions] =\ validate_and_internalize_conditions(step[var_names.block_if][var_names.conditions], srv_dict) # handle workflow if_workflow = [] for item in step[var_names.block_if][var_names.workflow]: if_workflow.append(validate_and_internalize_workflow_step( item, srv_dict, pol_dict, user_dict, team_dict, bus_srv_dict, cust_integ_dict, chat_integ_dict)) step[var_names.block_if][var_names.workflow] = if_workflow if var_names.block_else in step: else_workflow = [] for item in step[var_names.block_else]: else_workflow.append(validate_and_internalize_workflow_step( item, srv_dict, pol_dict, user_dict, team_dict, bus_srv_dict, cust_integ_dict, chat_integ_dict)) step[var_names.block_else] = else_workflow return step def validate_and_internalize_conditions(conditions, srv_dict): ''' Validates conditions and internalizes expected values. :param conditions: (list) of conditions in a workflow step :param srv_dict: (dict) -> {ref ID: service ID, ...} :return: (list) of internalized conditions :errors: AssertionError, KeyError, ValueError ''' allowed_and_or_options = [var_names.logic_and, var_names.logic_or] allowed_condition_params = [var_names.logic, var_names.rule_type, var_names.field_name, var_names.comparator, var_names.field_value] allowed_comparators = [constants.rule_contains, constants.rule_not_contains, constants.rule_equals, constants.rule_not_equals, constants.rule_exists, constants.rule_not_exists, constants.rule_matches, constants.rule_not_matches] assert isinstance(conditions, list) new_conditions = [] for grp in conditions: assert isinstance(grp, dict) and len(grp) == 1 grp_logic = list(grp.keys())[0] assert grp_logic in allowed_and_or_options new_grp = {grp_logic: []} grp_conds = grp[grp_logic] for cond in grp_conds: assert set(cond.keys()) == set(allowed_condition_params) rule_type = cond[var_names.rule_type] field_name = cond[var_names.field_name] field_value = cond[var_names.field_value] if rule_type not in [var_names.payload_field, var_names.workflow_not_run]: assert field_name == rule_type if rule_type != var_names.workflow_not_run: assert cond[var_names.comparator] in allowed_comparators new_cond = { var_names.logic: cond[var_names.logic], var_names.rule_type: rule_type, var_names.field_name: field_name, var_names.comparator: cond[var_names.comparator], var_names.field_value: field_value } if rule_type == var_names.service: unmasked_srv_key = key_manager.unmask_reference_key(field_value) if unmasked_srv_key in srv_dict: new_cond[var_names.field_value] = srv_dict[unmasked_srv_key] else: raise ValueError(errors.err_invalid_value) elif rule_type == var_names.status: assert field_value in [constants.open_state, constants.acknowledged_state, constants.resolved_state] elif rule_type == var_names.urgency_level: assert field_value in configs.allowed_urgency_levels elif rule_type in [var_names.email_from, var_names.email_to, var_names.previous_action_json]: assert not string_validator.is_empty_string(field_value) elif rule_type == var_names.previous_action_http_status: assert isinstance(field_value, int) elif rule_type == var_names.payload_field: assert not string_validator.is_empty_string(field_name) assert not string_validator.is_empty_string(field_value) elif rule_type == var_names.workflow_not_run: assert cond[var_names.comparator] is None assert cond[var_names.field_value] is None else: raise KeyError(errors.err_invalid_value) new_grp[grp_logic].append(new_cond) new_conditions.append(new_grp) return new_conditions def validate_and_internalize_workflow_action(action, pol_dict, user_dict, team_dict, bus_srv_dict, cust_integ_dict, chat_integ_dict): ''' Validates workflow action and internalizes the expected values where needed. :param action: (dict) -> {actions: {...}} :param pol_dict: (dict) -> {ref id: policy ID, ...} :param user_dict: (dict) -> {pref name: user ID, ...} :param team_dict: (dict) -> {ref id: team ID, ...} :param bus_srv_dict: (dict) -> {ref ID: business service ID, ...} :param cust_integ_dict: (dict) -> {integration key: integration ID, ...} :param chat_integ_dict: (dict) -> {integration key: integration ID, ...} :return: (dict) -> internalized actions dict :errors: AssertionError, KeyError, ValueError ''' assert isinstance(action, dict) assert len(action) == 1 assert isinstance(action[var_names.actions], dict) action_keys = list(action[var_names.actions].keys()) # Add new responders if var_names.new_responders in action_keys: assert len(action[var_names.actions]) == 1 new_assignees = action[var_names.actions][var_names.new_responders] assert isinstance(new_assignees, list) and len(new_assignees) > 0 intrz_resps = [] for pol_ref in new_assignees: unmasked_pol_ref = key_manager.unmask_reference_key(pol_ref) if unmasked_pol_ref in pol_dict: intrz_resps.append(pol_dict[unmasked_pol_ref]) else: raise ValueError(errors.err_invalid_value) action[var_names.actions][var_names.new_responders] = intrz_resps # Subscribe users elif var_names.user_subscribers in action_keys: assert len(action[var_names.actions]) == 1 new_users = action[var_names.actions][var_names.user_subscribers] assert isinstance(new_users, list) and len(new_users) > 0 action[var_names.actions][var_names.user_subscribers] = [user_dict[x] for x in new_users if x in user_dict] # Subscribe teams elif var_names.team_subscribers in action_keys: assert len(action[var_names.actions]) == 1 new_teams = action[var_names.actions][var_names.team_subscribers] assert isinstance(new_teams, list) and len(new_teams) > 0 intrz_teams = [] for tm_ref in new_teams: unmasked_tm_ref = key_manager.unmask_reference_key(tm_ref) if unmasked_tm_ref in team_dict: intrz_teams.append(team_dict[unmasked_tm_ref]) action[var_names.actions][var_names.team_subscribers] = intrz_teams # Add conference bridge elif var_names.conference_bridge in action_keys: assert len(action[var_names.actions]) == 1 assert isinstance(action[var_names.actions][var_names.conference_bridge], int) # Post a status update elif var_names.status_update in action_keys: assert len(action[var_names.actions]) == 1 sts_upd = action[var_names.actions][var_names.status_update] assert isinstance(sts_upd, str) and not string_validator.is_empty_string(sts_upd) # Add a note elif var_names.notes in action_keys: assert len(action[var_names.actions]) == 1 txt_note = action[var_names.actions][var_names.notes] assert isinstance(txt_note, str) and not string_validator.is_empty_string(txt_note) # Add a business impact elif var_names.business_impact in action_keys: assert len(action[var_names.actions]) == 1 new_imp = action[var_names.actions][var_names.business_impact] assert isinstance(new_imp, list) and len(new_imp) > 0 intrz_bus = [] for bus_ref in new_imp: unmasked_bus_ref = key_manager.unmask_reference_key(bus_ref) if unmasked_bus_ref in bus_srv_dict: intrz_bus.append(bus_srv_dict[unmasked_bus_ref]) action[var_names.actions][var_names.business_impact] = intrz_bus elif var_names.reassign_to in action_keys: assert len(action[var_names.actions]) == 1 new_assignees = action[var_names.actions][var_names.reassign_to] assert isinstance(new_assignees, list) and len(new_assignees) > 0 intrz_pols = [] for pol_ref in new_assignees: unmasked_pol_ref = key_manager.unmask_reference_key(pol_ref) if unmasked_pol_ref in pol_dict: intrz_pols.append(pol_dict[unmasked_pol_ref]) else: raise ValueError(errors.err_invalid_value) action[var_names.actions][var_names.reassign_to] = intrz_pols elif var_names.to_resolve in action_keys: assert len(action[var_names.actions]) == 1 assert isinstance(action[var_names.actions][var_names.to_resolve], bool) elif var_names.send_chat_message in action_keys: assert len(action[var_names.actions]) == 2 unmasked_integ_key = key_manager.unmask_reference_key(action[var_names.actions][var_names.send_chat_message]) if unmasked_integ_key in chat_integ_dict: action[var_names.actions][var_names.send_chat_message] = chat_integ_dict[unmasked_integ_key] else: raise ValueError(errors.err_invalid_value) txt_msg = action[var_names.actions][var_names.text_msg] assert not string_validator.is_empty_string(txt_msg) elif var_names.email in action_keys: assert len(action[var_names.actions]) == 3 email_to = action[var_names.actions][var_names.email] email_subj = action[var_names.actions][var_names.email_subject] txt_msg = action[var_names.actions][var_names.text_msg] for addr in email_to: assert string_validator.is_email_address(addr) assert not string_validator.is_empty_string(email_subj) assert not string_validator.is_empty_string(txt_msg) elif var_names.send_sms in action_keys: assert len(action[var_names.actions]) == 2 sms_to = action[var_names.actions][var_names.send_sms] assert isinstance(sms_to, list) txt_msg = action[var_names.actions][var_names.text_msg] assert not string_validator.is_empty_string(txt_msg) elif var_names.integration_id in action_keys: assert len(action[var_names.actions]) == 1 unmasked_integ_key = key_manager.unmask_reference_key(action[var_names.actions][var_names.integration_id]) if unmasked_integ_key in cust_integ_dict: action[var_names.actions][var_names.integration_id] = cust_integ_dict[unmasked_integ_key] else: raise ValueError(errors.err_invalid_value) else: raise KeyError(errors.err_invalid_value) return action def externalize_workflow(conn, timestamp, org_id, workflow_steps): ''' Externalize workflows. Switch internal IDs to an external ID. :param conn: db connection :param timestamp: timestamp when the request is being made :param org_id: ID of the organization :param workflow_steps: (list) of workflow steps :return: (list) of workflow steps ''' srv_dict = {v: k for k, v in db_services.list_service_ids_from_ref_ids( conn, timestamp, org_id, as_dict=True).items()} pol_dict = {v: k for k, v in db_policies.list_policy_ids_from_ref_ids( conn, timestamp, org_id, as_dict=True).items()} user_dict = {v: k for k, v in db_users.get_user_ids_from_preferred_usernames( conn, timestamp, org_id, as_dict=True).items()} team_dict = {v: k for k, v in db_teams.get_team_ids_from_ref_ids( conn, timestamp, org_id, as_dict=True).items()} bus_srv_dict = {v: k for k, v in db_business_services.get_business_service_ids_from_ref_ids( conn, timestamp, org_id).items()} cust_integ_dict = {v: k for k, v in db_integrations.list_custom_action_integrations_from_keys( conn, timestamp, org_id, as_dict=True).items()} chat_integ_dict = {v: k for k, v in db_integrations.list_chat_integrations_from_keys( conn, timestamp, org_id, as_dict=True).items()} new_steps = [] for step in workflow_steps: new_steps.append(externalize_workflow_step( step, srv_dict, pol_dict, user_dict, team_dict, bus_srv_dict, cust_integ_dict, chat_integ_dict)) return new_steps def externalize_workflow_step(step, srv_dict, pol_dict, user_dict, team_dict, bus_srv_dict, cust_integ_dict, chat_integ_dict): ''' Externalize a single workflow step. :param step: (dict) single workflow step :param srv_dict: (dict) -> {service ID: ref ID, ...} :param pol_dict: (dict) -> {policy ID: ref ID, ...} :param user_dict: (dict) -> {user ID: pref name, ...} :param team_dict: (dict) -> {team ID: ref ID, ...} :param bus_srv_dict: (dict) -> {bus ID: ref ID, ...} :param cust_integ_dict: (dict) -> {integration ID: integration key, ...} :param chat_integ_dict: (dict) -> {integration ID: integration key, ...} :return: (dict) -> externalize workflow ''' if var_names.actions in step: step = externalize_workflow_action(step, pol_dict, user_dict, team_dict, bus_srv_dict, cust_integ_dict, chat_integ_dict) else: if var_names.block_if in step: # handle conditions step[var_names.block_if][var_names.conditions] =\ externalize_conditions(step[var_names.block_if][var_names.conditions], srv_dict) # handle workflow if_workflow = [] for item in step[var_names.block_if][var_names.workflow]: if_workflow.append(externalize_workflow_step( item, srv_dict, pol_dict, user_dict, team_dict, bus_srv_dict, cust_integ_dict, chat_integ_dict)) step[var_names.block_if][var_names.workflow] = if_workflow if var_names.block_else in step: else_workflow = [] for item in step[var_names.block_else]: else_workflow.append(externalize_workflow_step( item, srv_dict, pol_dict, user_dict, team_dict, bus_srv_dict, cust_integ_dict, chat_integ_dict)) step[var_names.block_else] = else_workflow return step def externalize_conditions(conditions, srv_dict): ''' Externalize the conditions of a workflow. :param conditions: (list) of conditions :param srv_dict: (dict) -> {service ID: ref id, ...} :return: (list) of conditions ''' new_conditions = [] for grp in conditions: grp_logic = list(grp.keys())[0] new_grp = {grp_logic: []} grp_conds = grp[grp_logic] for cond in grp_conds: field_value = cond[var_names.field_value] if cond[var_names.rule_type] == var_names.service: if field_value in srv_dict: cond[var_names.field_value] = key_manager.conceal_reference_key(srv_dict[field_value]) new_grp[grp_logic].append(cond) new_conditions.append(new_grp) return new_conditions def externalize_workflow_action(action, pol_dict, user_dict, team_dict, bus_srv_dict, cust_integ_dict, chat_integ_dict): ''' Externalize workflow actions. :param action: (dict) -> {actions: {...}} :param pol_dict: (dict) -> {policy ID: ref ID, ...} :param user_dict: (dict) -> {user ID: pref name, ...} :param team_dict: (dict) -> {team ID: ref ID, ...} :param bus_srv_dict: (dict) -> {bus ID: ref ID, ...} :param cust_integ_dict: (dict) -> {integration ID: integration key, ...} :param chat_integ_dict: (dict) -> {integration ID: integration key, ...} :return: (dict) -> externalize actions ''' action_keys = list(action[var_names.actions].keys()) if var_names.new_responders in action_keys: ext_resps = [] for pol_id in action[var_names.actions][var_names.new_responders]: if pol_id in pol_dict: ext_resps.append(key_manager.conceal_reference_key(pol_dict[pol_id])) action[var_names.actions][var_names.new_responders] = ext_resps elif var_names.user_subscribers in action_keys: new_users = action[var_names.actions][var_names.user_subscribers] action[var_names.actions][var_names.user_subscribers] = [user_dict[x] for x in new_users if x in user_dict] elif var_names.team_subscribers in action_keys: ext_teams = [] for tm_id in action[var_names.actions][var_names.team_subscribers]: if tm_id in team_dict: ext_teams.append(key_manager.conceal_reference_key(team_dict[tm_id])) action[var_names.actions][var_names.team_subscribers] = ext_teams elif var_names.business_impact in action_keys: ext_bus = [] for bus_id in action[var_names.actions][var_names.business_impact]: if bus_id in bus_srv_dict: ext_bus.append(key_manager.conceal_reference_key(bus_srv_dict[bus_id])) action[var_names.actions][var_names.business_impact] = ext_bus elif var_names.reassign_to in action_keys: ext_resps = [] for pol_id in action[var_names.actions][var_names.reassign_to]: if pol_id in pol_dict: ext_resps.append(key_manager.conceal_reference_key(pol_dict[pol_id])) action[var_names.actions][var_names.reassign_to] = ext_resps elif var_names.send_chat_message in action_keys: integ_id = action[var_names.actions][var_names.send_chat_message] if integ_id in chat_integ_dict: action[var_names.actions][var_names.send_chat_message] =\ key_manager.conceal_reference_key(chat_integ_dict[integ_id]) elif var_names.integration_id in action_keys: integ_id = action[var_names.actions][var_names.integration_id] if integ_id in cust_integ_dict: action[var_names.actions][var_names.integration_id] =\ key_manager.conceal_reference_key(cust_integ_dict[integ_id]) return action def check_component_in_workflow(conn, timestamp, organization_id, pol_id=None, replacing_pol_id=None, user_id=None, replacing_user_id=None, team_id=None, srv_id=None, bus_id=None, integ_id=None, conf_id=None): ''' Checks component dependencies in all workflows of an organization. :param conn: db connection :param timestamp: timestamp when this request is being made :param organization_id: ID of the organization whose workflows should be checked :param pol_id: policy ID to check for :param replacing_pol_id: policy ID to replace with :param user_id: user ID to check for :param replacing_user_id: user ID to replace with :param team_id: ID of the team to check for :param srv_id: ID of the service to check for :param bus_id: ID of the business service to check for :param integ_id: ID of the custom action integration to check for :param conf_id: ID of the conference bridge to check for :return: (list) of updated workflows | when conflicts are found, a list of conflicting workflows are generated ''' updated_workflows = [] available_workflows = get_workflows(conn, timestamp, organization_id, with_ref_id_maps=False) for item in available_workflows: with_run_team = item[var_names.run_permissions] with_edit_team = item[var_names.edit_permissions] try: if team_id is not None: if (with_run_team is not None and with_run_team == team_id) or\ (with_edit_team is not None and with_edit_team == team_id): raise DependencyFound(errors.err_workflow_dependency_team_delete) old_steps = copy.deepcopy(item[var_names.workflow]) new_steps = [] for step in item[var_names.workflow]: new_steps.append(check_component_in_workflow_step( step, pol_id, replacing_pol_id, user_id, replacing_user_id, team_id, srv_id, bus_id, integ_id, conf_id )) if old_steps != new_steps: updated_workflows.append({ var_names.workflow_id: item[var_names.workflow_id], var_names.workflow_name: item[var_names.workflow_name], var_names.workflow: new_steps }) except DependencyFound: updated_workflows.append({ var_names.workflow_id: item[var_names.workflow_id], var_names.workflow_name: item[var_names.workflow_name], var_names.workflow: item[var_names.workflow] }) return updated_workflows def check_component_in_workflow_step(step, pol_id=None, replacing_pol_id=None, user_id=None, replacing_user_id=None, team_id=None, srv_id=None, bus_id=None, integ_id=None, conf_id=None): ''' Check component dependency inside a single workflow step. :param step: (dict) single workflow step :param pol_id: policy ID to check for :param replacing_pol_id: policy ID to replace with :param user_id: user ID to check for :param replacing_user_id: user ID to replace with :param team_id: ID of the team to check for :param srv_id: ID of the service to check for :param bus_id: ID of the business service to check for :param integ_id: ID of the custom action integration to check for :param conf_id: ID of the conference bridge to check for :return: (dict) -> internalized workflow step :error: AssertionError, DatabaseError, KeyError, ValueError ''' if var_names.actions in step: step = check_component_in_workflow_action(step, pol_id, replacing_pol_id, user_id, replacing_user_id, team_id, bus_id, integ_id, conf_id) else: if var_names.block_if in step: if srv_id is not None: check_component_in_conditions(step[var_names.block_if][var_names.conditions], srv_id) # handle workflow if_workflow = [] for item in step[var_names.block_if][var_names.workflow]: if_workflow.append(check_component_in_workflow_step( item, pol_id, replacing_pol_id, user_id, replacing_user_id, team_id, bus_id, integ_id, conf_id)) step[var_names.block_if][var_names.workflow] = if_workflow if var_names.block_else in step: else_workflow = [] for item in step[var_names.block_else]: else_workflow.append(check_component_in_workflow_step( item, pol_id, replacing_pol_id, user_id, replacing_user_id, team_id, bus_id, integ_id, conf_id)) step[var_names.block_else] = else_workflow return step def check_component_in_conditions(conditions, srv_id): ''' Check component dependency inside workflow conditions. :param conditions: (list) of conditions in a workflow step :param srv_id: (dict) -> ID of the service to check for :errors: AssertionError, KeyError, ValueError ''' for grp in conditions: grp_logic = list(grp.keys())[0] grp_conds = grp[grp_logic] for cond in grp_conds: if cond[var_names.rule_type] == var_names.service and cond[var_names.field_value] == srv_id: raise DependencyFound(errors.err_workflow_dependency_service_delete) def check_component_in_workflow_action(action, pol_id=None, replacing_pol_id=None, user_id=None, replacing_user_id=None, team_id=None, bus_id=None, integ_id=None, conf_id=None): ''' Check component dependency inside workflow actions. :param action: (dict) -> {actions: {...}} :param pol_id: policy ID to check for :param replacing_pol_id: policy ID to replace with :param user_id: user ID to check for :param replacing_user_id: user ID to replace with :param team_id: ID of the team to check for :param bus_id: ID of the business service to check for :param integ_id: ID of the custom action integration to check for :param conf_id: ID of the conference bridge to check for :return: (dict) -> internalized actions dict :errors: DependencyFound ''' action_keys = list(action[var_names.actions].keys()) if pol_id is not None: if var_names.new_responders in action_keys: new_assignees = action[var_names.actions][var_names.new_responders] if pol_id in new_assignees: new_assignees.remove(pol_id) if replacing_pol_id is None: if len(new_assignees) == 0: raise DependencyFound(errors.err_workflow_dependency_policy_delete) else: if replacing_pol_id not in new_assignees: new_assignees.append(replacing_pol_id) action[var_names.actions][var_names.new_responders] = new_assignees elif var_names.reassign_to in action_keys: new_assignees = action[var_names.actions][var_names.reassign_to] if pol_id in new_assignees: new_assignees.remove(pol_id) if replacing_pol_id is None: if len(new_assignees) == 0: raise DependencyFound(errors.err_workflow_dependency_policy_delete) else: if replacing_pol_id not in new_assignees: new_assignees.append(replacing_pol_id) action[var_names.actions][var_names.reassign_to] = new_assignees if user_id is not None: if var_names.user_subscribers in action_keys: new_users = action[var_names.actions][var_names.user_subscribers] if user_id in new_users: new_users.remove(user_id) if replacing_user_id is None: if len(new_users) == 0: raise DependencyFound(errors.err_workflow_dependency_component_delete) else: if replacing_user_id not in new_users: new_users.append(replacing_user_id) action[var_names.actions][var_names.user_subscribers] = new_users if team_id is not None: if var_names.team_subscribers in action_keys\ and team_id in action[var_names.actions][var_names.team_subscribers]: raise DependencyFound(errors.err_workflow_dependency_team_delete) if conf_id is not None: if var_names.conference_bridge in action_keys\ and action[var_names.actions][var_names.conference_bridge] == conf_id: raise DependencyFound(errors.err_workflow_dependency_conference_bridge_delete) if bus_id is not None: if var_names.business_impact in action_keys and bus_id in action[var_names.actions][var_names.business_impact]: raise DependencyFound(errors.err_workflow_dependency_component_delete) if integ_id is not None: if ((var_names.integration_id in action_keys and integ_id in action[var_names.actions][var_names.integration_id]) or (var_names.send_chat_message in action_keys and integ_id in action[var_names.actions][var_names.send_chat_message])): raise DependencyFound(errors.err_workflow_dependency_component_delete) return action def list_workflow_step_conditions_and_actions(step, is_conditional, actions_list): ''' Externalize a single workflow step. :param step: (dict) single workflow step :param is_conditional: boolean that states if the workflow has conditional statements or not :param actions_list: (list) of actions to perform :return: (tuple) -> is conditional, list of actions ''' if var_names.actions in step: actions_list += list(step[var_names.actions].keys()) else: if var_names.block_if in step: if not is_conditional: is_conditional = True for item in step[var_names.block_if][var_names.workflow]: is_conditional, actions_list = list_workflow_step_conditions_and_actions( item, is_conditional, actions_list) if var_names.block_else in step: for item in step[var_names.block_else]: is_conditional, actions_list = list_workflow_step_conditions_and_actions( item, is_conditional, actions_list) return is_conditional, actions_list