# By: Riasat Ullah # This file contains functions that can be used to validate the information # provided to create/edit configurations related entries in the database. from utils import constants, errors, key_manager, times, var_names from validations import string_validator import configuration as configs import pytz def validate_conditional_routing_data(is_enabled, rule_name, rule_description, valid_start_date, valid_end_date, valid_start_time, valid_end_time, repeat_on, rule_application_count, rules, actions, timezone, allow_multiple): ''' Validate data provided to create/edit a conditional routing directive. :param is_enabled: (boolean) True if the routing directive is enabled; False otherwise :param rule_name: the name/title of the conditional routing directive :param rule_description: (optional) description of what the conditional routing is for :param valid_start_date: the date from when this routing will apply :param valid_end_date: the date this routing will apply till :param valid_start_time: the time in the day this routing will apply from :param valid_end_time: the time in the day this routing will apply till :param repeat_on: the days the routing will apply on :param rule_application_count: the number of rules that must apply (>= 1) :param rules: the conditional routing rules :param actions: the actions to perform :param timezone: the timezone this routing will apply in; important for start and end dates and times :param allow_multiple: allow other conditional routing rules to be combined :errors: AssertionError, ValueError ''' assert isinstance(is_enabled, bool) assert string_validator.is_standard_name(rule_name) if rule_description is not None: assert isinstance(rule_description, str) if allow_multiple is not None: assert isinstance(allow_multiple, bool) # check that valid start and end dates and times are in the correct format assert times.get_date_from_string(valid_start_date) < times.get_date_from_string(valid_end_date) try: start_ = times.get_time_from_string(valid_start_time) end_ = times.get_time_from_string(valid_end_time) if start_ > end_: raise ValueError(errors.err_time_end_before_start) except (AssertionError, TypeError): raise AssertionError(errors.err_internal_time_format) # validate repeat days if len(repeat_on) > 0: assert isinstance(repeat_on, list) assert set(repeat_on).issubset(set(range(0, 7))) # at least 1 rule must apply assert isinstance(rule_application_count, int) and 1 <= rule_application_count <= len(rules) # check the timezone assert timezone in pytz.all_timezones validate_routing_rules(rules) validate_routing_actions(actions) def validate_routing_rules(rules): ''' Validate the routing rules that are sent as data for creating/editing conditional routing directives. :param rules: (list) of dict containing routing rules :errors: AssertionError ''' rule_attributes = {var_names.rule_type, var_names.field_name, var_names.comparator, var_names.field_value} rule_comparators = [constants.rule_contains, constants.rule_equals, constants.rule_exists, constants.rule_matches, constants.rule_not_contains, constants.rule_not_equals, constants.rule_not_exists, constants.rule_not_matches] rule_types = [var_names.email_from, var_names.email_to, var_names.email_subject, var_names.payload, var_names.service, var_names.urgency_level] for item in rules: assert isinstance(item, dict) assert set(item.keys()) == rule_attributes assert item[var_names.rule_type] in rule_types assert isinstance(item[var_names.field_name], str) assert item[var_names.comparator] in rule_comparators if item[var_names.rule_type] == var_names.service: # make sure the service ref id provided is concealed key_manager.unmask_reference_key(item[var_names.field_value]) elif item[var_names.rule_type] == var_names.urgency_level: assert item[var_names.field_value] in configs.allowed_urgency_levels else: assert isinstance(item[var_names.field_value], str) def validate_routing_actions(actions): ''' Validate the actions that are expected to be performed if the conditional routing directives are met. :param actions: (dict) of actions :errors: AssertionError ''' # All the action attributes do not have to be included. Only include the ones that apply. action_attributes = {var_names.alert_handling, var_names.route_to, var_names.route_to_service, var_names.resolve_incidents, var_names.resolve_hours, var_names.suppress_count, var_names.suppress_minutes, var_names.suppress_before, var_names.group_minutes, var_names.urgency_level, var_names.tags, var_names.incident_title_field, var_names.incident_description_field, var_names.dedup_key_field, var_names.notes} assert isinstance(actions, dict) assert set(actions.keys()).issubset(action_attributes) # alert handling is a mandatory attribute assert var_names.alert_handling in actions handler = actions[var_names.alert_handling] assert handler in [constants.normal_alert, constants.resolve_alert, constants.cancel_alert, constants.re_route_alert, constants.re_route_service] if handler == constants.re_route_alert: assert var_names.route_to in actions # Make sure the policy ref id is concealed. Internalization is not handled here. It has to be handled later. try: key_manager.unmask_reference_key(actions[var_names.route_to]) except TypeError: raise AssertionError(errors.err_invalid_request) else: assert var_names.route_to not in actions if handler == constants.re_route_service: assert var_names.route_to_service in actions # Make sure the service ref id is concealed. Internalization is not handled here. It has to be handled later. try: key_manager.unmask_reference_key(actions[var_names.route_to_service]) except TypeError: raise AssertionError(errors.err_invalid_request) else: assert var_names.route_to_service not in actions if handler == constants.resolve_alert: if var_names.resolve_incidents in actions: assert isinstance(actions[var_names.resolve_incidents], bool) assert var_names.resolve_hours in actions and actions[var_names.resolve_hours] > 0 # alert suppressing, grouping and urgency level updating if var_names.suppress_count in actions: assert isinstance(actions[var_names.suppress_count], int) assert var_names.suppress_minutes in actions if var_names.suppress_minutes in actions: assert isinstance(actions[var_names.suppress_minutes], int) and actions[var_names.suppress_minutes] > 0 assert var_names.suppress_count in actions if var_names.suppress_before in actions: assert isinstance(actions[var_names.suppress_before], bool) assert var_names.suppress_count in actions and var_names.suppress_minutes in actions if var_names.group_minutes in actions: assert isinstance(actions[var_names.group_minutes], int) and actions[var_names.group_minutes] > 0 if var_names.urgency_level in actions: assert actions[var_names.urgency_level] in configs.allowed_urgency_levels if var_names.tags in actions: for tag_item in actions[var_names.tags]: assert isinstance(tag_item, str) # customization, field extraction and notes if var_names.incident_title_field in actions: assert isinstance(actions[var_names.incident_title_field], str) if var_names.incident_description_field in actions: assert isinstance(actions[var_names.incident_description_field], str) if var_names.dedup_key_field in actions: assert isinstance(actions[var_names.dedup_key_field], str) if var_names.notes in actions: assert isinstance(actions[var_names.notes], str) def validate_live_call_routing_data(routing_name, routing_description, greeting_text, greeting_audio_filename, greeting_audio_location, greeting_audio_url, ending_text, ending_audio_filename, ending_audio_location, ending_audio_url, default_service, option_services, text_language, male_voice, max_forwarding_users, forwarding_timeout, incident_urgency, block_numbers, show_caller_id, resolve_answered_calls, resolve_unanswered_calls, to_alert, record_voice_mail, record_call, prompt_call_acceptance, prompt_format, incident_title_format): ''' Validate the customer entered live call routing data. :param routing_name: short name for the live call routing specification :param routing_description: description for the live call routing :param greeting_text: the greeting text which will be converted to speech and played when the user calls :param greeting_audio_filename: (str) filename of the greeting audio (only if greeting text not provided) :param greeting_audio_location: (str) s3 key location of the greeting audio (only if greeting text not provided) :param greeting_audio_url: (str) http url to fetch the audio with (only if greeting text no provided) :param ending_text: the ending note which will be converted to speech and played before ending the call :param ending_audio_filename: (str) filename of the ending audio (only if ending text not provided) :param ending_audio_location: (str) s3 key location of the ending audio (only if ending text not provided) :param ending_audio_url: (str) http url to fetch the audio with (only if ending text not provided) :param default_service: (int) ID of the service the call will be forwarded to by default :param option_services: (dict) options mapped to forwarding service IDs - {option: service ID, ...} :param text_language: (str) language the greeting and ending texts are in :param male_voice: (boolean) True if the text to speech voice should be male; False otherwise :param max_forwarding_users: (int) maximum number of users who the call should be forwarded to :param forwarding_timeout: (int) number of seconds after which each forwarding attempt will be ended :param incident_urgency: (int) urgency the created incident will be set to :param block_numbers: (list of str) list of phone numbers to block or disallow :param show_caller_id: (boolean) True if the incoming caller's phone number should be shown when forwarding to on-call users or the routing phone number should be shown :param resolve_answered_calls: (boolean) True if the incidents should be auto-resolved if the call is answered :param resolve_unanswered_calls: (boolean) True if the incidents should be auto-resolved if the call is not answered :param to_alert: (boolean) True if alert notifications should be sent out if the incident remains unresolved :param record_voice_mail: (boolean) True if voice mail should be prompted and recorded :param record_call: (boolean) True if the call should be recorded :param prompt_call_acceptance: (boolean) True if the responder should be prompted to accept the call :param prompt_format: (str) format of the prompt to play :param incident_title_format: (str) format of the title to create the incident with :errors: AssertionError ''' assert string_validator.is_standard_name(routing_name) if routing_description is not None: assert isinstance(routing_description, str) # Greeting and ending notes are not mandatory, but if they are provided # then either the text or audio file must be given. if greeting_text is not None: assert isinstance(greeting_text, str) if greeting_audio_filename is not None: assert greeting_audio_filename is not None and isinstance(greeting_audio_filename, str) if greeting_audio_location is not None or greeting_audio_url is not None: assert greeting_audio_location is not None and string_validator.is_file_path(greeting_audio_location) assert greeting_audio_url is not None and string_validator.is_web_url(greeting_audio_url) if ending_text is not None: assert isinstance(ending_text, str) if ending_audio_filename is not None: assert isinstance(ending_audio_filename, str) if ending_audio_location is not None or ending_audio_url is not None: assert ending_audio_location is not None and string_validator.is_file_path(ending_audio_location) assert ending_audio_url is not None and string_validator.is_web_url(ending_audio_url) # Forwarding services key_manager.unmask_reference_key(default_service) if option_services is not None: assert isinstance(option_services, dict) assert len(option_services) <= configs.call_routing_max_options for key in option_services: int(key) key_manager.unmask_reference_key(option_services[key]) # routing settings assert text_language in configs.allowed_live_call_routing_languages assert isinstance(male_voice, bool) if max_forwarding_users is not None: assert isinstance(max_forwarding_users, int) if forwarding_timeout is not None: assert isinstance(forwarding_timeout, int) assert incident_urgency in configs.allowed_urgency_levels if block_numbers is not None: assert isinstance(block_numbers, list) for item in block_numbers: assert string_validator.is_phone_number(item) assert isinstance(show_caller_id, bool) assert isinstance(resolve_answered_calls, bool) assert isinstance(resolve_unanswered_calls, bool) assert isinstance(to_alert, bool) assert isinstance(record_voice_mail, bool) assert isinstance(record_call, bool) assert isinstance(prompt_call_acceptance, bool) assert isinstance(prompt_format, str) assert isinstance(incident_title_format, str) def validate_check_data(check_ref_id, check_type, check_name, check_description, ping_type, ping_url, ping_email, ping_interval, grace_period, service_ref_id, inc_title, inc_description, urgency_level, tags, ip_whitelist, email_whitelist, packet_sizes, degraded_timeout, fail_timeout, check_ssl_expiry, ping_port, additional_info): ''' Validate monitor check data. @param check_ref_id: reference ID of the check; only for heartbeat checks this will be pre-generated and provided with the user request @param check_type: type of check (HEARTBEAT, etc) @param check_name: name for the check @param check_description: short description of what the check is for @param ping_type: type of ping method that will be used for the check (HTTP, EMAIL) @param ping_url: (for HTTP pings) url of the ping endpoint @param ping_email: (for email pings) email of the ping endpoint @param ping_interval: (integer) interval between pings @param grace_period: grace period allowed for successful pings @param service_ref_id: reference ID of the service this check is for @param inc_title: title of the incident that should be created if the check fails @param inc_description: description of the incident that should be created if the check fails @param urgency_level: urgency of the incident to be created if the check fails @param tags: (list of str) tags to be added to the incident that will be created if the check fails @param ip_whitelist: (list of IP addresses) IP addresses that HTTP requests will be allowed from @param email_whitelist: (list of email) email addresses that email pings will be allowed from @param packet_sizes: (list of int) packet sizes in kilobytes that should be used for the check @param degraded_timeout: (int) number of seconds to allow before a ping response is considered degraded @param fail_timeout: (int) number of seconds to allow before a ping response is considered to have failed @param check_ssl_expiry: (boolean) true if domain SSL expiry should be checked @param ping_port: (int) port to ping @param additional_info: (dict) additional information necessary for the check ''' # validate the basic ping information assert check_type in configs.allowed_monitor_checks assert string_validator.is_standard_name(check_name) if check_description is not None: assert isinstance(check_description, str) if check_type == constants.heartbeat: key_manager.unmask_reference_key(check_ref_id) assert ping_type in [constants.http, constants.email] else: assert check_ref_id is None assert ping_type == constants.http if ping_type == constants.http: assert string_validator.is_web_url(ping_url) or string_validator.is_valid_ip_address(ping_url) elif ping_type == constants.email: assert string_validator.is_email_address(ping_email) # Ping interval must be greater than 30s assert isinstance(ping_interval, int) and ping_interval >= 30 if grace_period is not None: assert isinstance(grace_period, int) # Service reference ID should come to the back end in concealed form. Unmask for validation. key_manager.unmask_reference_key(service_ref_id) # validate the details of the incident that should be created if the check fails assert isinstance(inc_title, str) and len(inc_title) > 0 if inc_description is not None: assert isinstance(inc_description, str) assert urgency_level in configs.allowed_urgency_levels if tags is not None: assert isinstance(tags, list) for tg in tags: assert isinstance(tg, str) # validate permissions if ip_whitelist is not None: assert isinstance(ip_whitelist, list) for ip in ip_whitelist: assert string_validator.is_valid_ip_address(ip) if email_whitelist is not None: assert isinstance(email_whitelist, list) for eml in email_whitelist: assert string_validator.is_email_address(eml) # validate optional ping parameters if packet_sizes is not None: assert isinstance(packet_sizes, list) for item in packet_sizes: assert isinstance(item, int) if degraded_timeout is not None: assert isinstance(degraded_timeout, int) if fail_timeout is not None: assert isinstance(fail_timeout, int) if check_ssl_expiry is not None: assert isinstance(check_ssl_expiry, bool) if ping_port is not None: assert isinstance(ping_port, int) if additional_info is not None: assert isinstance(additional_info, dict)