# By: Riasat Ullah # This file contains functions that help to validate Group related data. from utils import errors, key_manager, permissions, roles, times, var_names from validations import string_validator import configuration import datetime import pytz def validate_service_data(timestamp, org_id, service_name, description, re_trigger_minutes, support_days, support_start, support_end, support_timezone, de_prioritize, re_prioritize, allow_grouping): ''' Validate the format of the data to be entered for a service. Policy ref ID is not validated here as it gets unmasked directly from where this function gets called. :param timestamp: timestamp when this request is being made :param org_id: ID of the organization the service is for :param service_name: name of the service :param description: minor description of what the service is for :param re_trigger_minutes: number of minutes after which acknowledged incidents should be re-triggered :param support_days: days of the week the service should be active for support :param support_start: (str) the start time of the support period in the format HH:MM :param support_end: (str) the end time of the support period in the format HH:MM :param support_timezone: timezone the support hours are in :param de_prioritize: (boolean) to de-prioritize incidents during off hours or not :param re_prioritize: (boolean) to re-prioritize incidents that were de-prioritized when support hours start or not :param allow_grouping: (boolean) to allow incidents to be grouped on this service or not :errors: AssertionError, ValueError ''' assert isinstance(timestamp, datetime.datetime) assert isinstance(org_id, int) assert string_validator.is_standard_name(service_name) assert string_validator.is_not_sql_injection(description) if re_trigger_minutes is not None: # re-trigger minutes cannot be negative assert isinstance(re_trigger_minutes, int) assert re_trigger_minutes >= 0 if support_timezone is not None: assert support_timezone in pytz.all_timezones if de_prioritize is not None: assert isinstance(de_prioritize, bool) if re_prioritize is not None: assert isinstance(re_prioritize, bool) if allow_grouping is not None: assert isinstance(allow_grouping, bool) if support_days is not None: assert all(isinstance(x, int) and 0 <= x <= 6 for x in support_days) try: start_time = times.get_time_from_string(support_start) end_time = times.get_time_from_string(support_end) # the service support period must start before it ends if start_time > end_time: raise ValueError(errors.err_time_end_before_start) except (TypeError, ValueError): raise AssertionError(errors.err_internal_time_format) def validate_policy_data(timestamp, org_id: int, policy_name: str, policy_levels: list, org_routines: list): ''' Validates the information of a policy before entering it into the database. This validation is primarily used before creating or editing a policy. :param timestamp: (datetime.datetime) timestamp when the group was created :param org_id: ID of the organization the policy belongs to :param policy_name: (string) the name of the policy :param policy_levels: (list) of dict of policy levels data :param org_routines: (list of list) -> [ [routine name, concealed routine ref], ... ] :errors: SqlInjection, AssertionError, DatabaseError, KeyError ''' # Stakeholders cannot have routines. This is why we don't need to check that user's whose routines are added # have the right to be included in components. assert isinstance(timestamp, datetime.datetime) assert isinstance(org_id, int) assert string_validator.is_standard_name(policy_name) # The group must have at least 1 routine assert len(policy_levels) > 0 level_attributes = {var_names.assignee_level, var_names.level_minutes, var_names.routines} for item in policy_levels: # check that each level contains the required attributes item_attributes = item.keys() assert len(item_attributes) == len(level_attributes) and set(item_attributes) == level_attributes assert isinstance(item[var_names.assignee_level], int) assert isinstance(item[var_names.level_minutes], int) assert isinstance(item[var_names.routines], list) # make sure that the routine ids belong to the organization org_routine_refs = [item[1] for item in org_routines] # [routine ref 1, routine ref 2, ...] # the ref ids received here are already concealed; so we do not need to unmask them for the comparison for ref_id in item[var_names.routines]: assert ref_id in org_routine_refs def validate_routine_data(timestamp, organization_id, routine_name, routine_timezone, routine_layers, assignable_users): ''' Validates that the data provided for a single routine contains all the required information. :param timestamp: timestamp when this request is being made :param organization_id: ID of the organization :param routine_name: the name of the routine :param routine_timezone: the timezone the routine is for :param routine_layers: the layers of the routine :param assignable_users: (dict of list) -> { pref name: [user ID, user permission], ...} :errors: AssertionError ''' assert isinstance(timestamp, datetime.datetime) assert isinstance(organization_id, int) assert string_validator.is_not_sql_injection(routine_name) assert routine_timezone in pytz.all_timezones assert isinstance(routine_layers, list) all_rotations = [] for item in routine_layers: validate_routine_layer_data(item) all_rotations += item[var_names.rotations] validate_routine_rotations_data(all_rotations, assignable_users) def validate_routine_layer_data(data): ''' Validates a routine layer's data. :param data: (dict) layer's data :errors: AssertionError, ValueError ''' layer_attributes = {var_names.layer, var_names.layer_name, var_names.valid_start, var_names.valid_end, var_names.is_exception, var_names.rotation_start, var_names.shift_length, var_names.rotation_frequency, var_names.skip_days, var_names.rotations} assert isinstance(data, dict) # check that each routine layer contains the required attributes item_attributes = data.keys() assert len(item_attributes) == len(layer_attributes) and set(item_attributes) == layer_attributes # check the layer number, valid start, valid end, if the layer is an exception or not, # the rotation frequency, and if any days need to be skipped or not assert isinstance(data[var_names.layer], int) assert isinstance(data[var_names.layer_name], str) assert string_validator.is_date(data[var_names.valid_start]) if data[var_names.valid_end] is not None: assert string_validator.is_date(data[var_names.valid_end]) assert times.get_date_from_string(data[var_names.valid_start]) < times.get_date_from_string( data[var_names.valid_end]) assert isinstance(data[var_names.is_exception], bool) assert isinstance(data[var_names.rotation_frequency], int) if data[var_names.skip_days] is not None: assert isinstance(data[var_names.skip_days], list) for skip_day_num in data[var_names.skip_days]: assert isinstance(skip_day_num, int) and 0 <= skip_day_num <= 6 # check that routine start time and end time are in the correct format (HH:MM) and that the start is before the end try: times.get_time_from_string(data[var_names.rotation_start]) shift_len_list = data[var_names.shift_length].split(':') assert len(shift_len_list) == 2 assert 0 <= int(shift_len_list[0]) <= 24 assert 0 <= int(shift_len_list[1]) <= 59 if shift_len_list[0] == 24: assert int(shift_len_list[1]) == 0 except (TypeError, ValueError): raise AssertionError(errors.err_internal_time_format) def validate_routine_rotations_data(routine_rotations, assignable_users): ''' Validates routine layer rotations data; the data that is to be entered into the database. :param routine_rotations: (list) of rotation items of all the layers of the routine :param assignable_users: (dict of list) -> { pref name: [user ID, user permission], ...} :errors: AssertionError, DatabaseError, SqlInjection ''' # rotations is a list of [[preferred username(s) ...], ...] # -> [ [pref_name 1, pref_name 2], [pref_name 3], [pref_name 1, pref_name 4, pref_name 5], ... ] assignable_prefs = list(assignable_users.keys()) for rotation_list in routine_rotations: assert isinstance(rotation_list, list) for pref_name in rotation_list: if pref_name not in assignable_prefs: raise AssertionError(errors.err_unknown_resource + ' - ' + pref_name) user_perm = assignable_users[pref_name][1] if not permissions.has_user_permission(user_perm, permissions.USER_COMPONENTS_INCLUSION_PERMISSION): raise PermissionError(errors.err_component_inclusion) def validate_business_service_data(timestamp, organization_id, service_name, supporting_technical_services, supporting_business_services, associated_teams, description, min_urgency, org_tech_service_ref_ids, org_business_service_ref_ids, org_team_ref_ids): ''' Validates the data needed for a business service. :param timestamp: timestamp when this request is being made :param organization_id: ID of the organization this business service is for :param service_name: name of the business service :param supporting_technical_services: (list) of reference IDs of technical services (unmasked) :param supporting_business_services: (list) of reference IDs of business services (unmasked) :param associated_teams: (list) of IDs of teams (unmasked) :param description: brief description of what the business service is for :param min_urgency: minimum urgency of incidents triggered by supporting technical services that would impact this business service :param org_tech_service_ref_ids: (list) of concealed reference IDs of technical services of the organization :param org_business_service_ref_ids: (list) of concealed reference IDs of business services of the organization :param org_team_ref_ids: (list) of concealed reference IDs of teams of the organization :errors: AssertionError ''' assert isinstance(timestamp, datetime.datetime) assert isinstance(organization_id, int) assert string_validator.is_standard_name(service_name) if description is not None: assert string_validator.is_not_sql_injection(description) if min_urgency is not None: assert min_urgency in configuration.allowed_urgency_levels assert (supporting_technical_services is not None and len(supporting_technical_services) > 0) or\ (supporting_business_services is not None and len(supporting_business_services) > 0) if supporting_technical_services is not None: assert isinstance(supporting_technical_services, list) for ref_id in supporting_technical_services: assert ref_id in org_tech_service_ref_ids if supporting_business_services is not None: assert isinstance(supporting_business_services, list) for ref_id in supporting_business_services: assert ref_id in org_business_service_ref_ids if associated_teams is not None: assert isinstance(associated_teams, list) for ref_id in associated_teams: assert ref_id in org_team_ref_ids def validate_team_data(timestamp, organization_id, team_name, team_users, is_public, description): ''' Validates the data to be entered in the database for a team. :param timestamp: timestamp when this request is being made :param organization_id: ID of the organization the team is for :param team_name: name of the team :param team_users: (list of list) -> [ [user_id, team role], ...] :param is_public: (boolean) that states if the team should be publicly accessible within the organization :param description: brief description of what the team does :errors: AssertionError, ValueError ''' assert isinstance(timestamp, datetime.datetime) assert isinstance(organization_id, int) assert string_validator.is_standard_name(team_name) assert isinstance(is_public, bool) if description is not None: assert isinstance(description, str) assert isinstance(team_users, list) if len(team_users) == 0: raise ValueError(errors.err_team_member_count) # make sure that there are no duplicate user_ids all_user_ids = [x[0] for x in team_users] assert len(all_user_ids) == len(set(all_user_ids)) # create team member role permissions for item in team_users: assert isinstance(item, list) assert string_validator.is_valid_preferred_username(item[0]) assert item[1] in roles.advanced_component_roles def validate_integration_details(integ_type, integ_name, integ_email=None, integ_url=None, access_token=None, secret_token=None, vendor_endpoint=None, vendor_endpoint_name=None, vendor_account_name=None, incoming_events=None, outgoing_events=None, conditions_map=None, payload_map=None, public_access=None, additional_info=None, external_id=None, external_info=None, updated_by=None): ''' Validates the details of an integration. :param integ_type: (str) the type of the integration :param integ_name: user given name of the integration :param integ_email: email address of the integration if this is an email integration :param integ_url: url issued by TaskCall to receive integration messages into :param access_token: access token needed to exchange information for this integration :param secret_token: secret token needed to exchange information for this integration :param vendor_endpoint: url issued by the vendor for TaskCall to send messages to the vendor :param vendor_endpoint_name: name of the vendor's endpoint :param vendor_account_name: name of the organization's account at the vendor's :param incoming_events: (list) of incoming events that can be accepted through this integration :param outgoing_events: (list) of outgoing events that can be sent out through this integration :param conditions_map: (dict) of conditions that must be met before accepting messages :param payload_map: (dict) that maps the vendor fields and constants to TaskCall's internal variables :param public_access: (bool) that states if this integration can be connected to publicly :param additional_info: (dict) any extra information :param external_id: (str) ID of the organization on the vendor's side :param external_info: (dict) of external details of the organization :param updated_by: user_id of the user who created/updated the details :errors: AssertionError ''' assert integ_type in configuration.allowed_integration_types assert string_validator.is_standard_name(integ_name) if integ_email is not None: assert string_validator.is_email_address(integ_email) if integ_url is not None: assert string_validator.is_web_url(integ_url) if access_token is not None: assert isinstance(access_token, str) if secret_token is not None: assert isinstance(secret_token, str) if vendor_endpoint is not None: assert string_validator.is_web_url(vendor_endpoint) if vendor_endpoint_name is not None: assert isinstance(vendor_endpoint_name, str) if vendor_account_name is not None: assert isinstance(vendor_account_name, str) # these fields are for internal use only if incoming_events is not None: assert isinstance(incoming_events, list) assert set(incoming_events).issubset(set(configuration.allowed_integration_events)) if outgoing_events is not None: assert isinstance(outgoing_events, list) assert set(outgoing_events).issubset(set(configuration.allowed_integration_events)) if conditions_map is not None: assert isinstance(conditions_map, dict) if payload_map is not None: assert isinstance(payload_map, dict) if public_access is not None: assert isinstance(public_access, bool) if additional_info is not None: assert isinstance(additional_info, dict) if external_id is not None: assert isinstance(external_id, str) if external_info is not None: assert isinstance(external_info, dict) if updated_by is not None: assert isinstance(updated_by, int) def service_circular_dependency(check, start, service_map, circular=None): ''' Gets the circular dependency of services (both technical and business services). The data for this check has to come from the internal tech and business dependency queries in the db_services file. :param check: ID of the dependency to check :param start: ID of the dependency to start from :param service_map: (dict) -> {service id: {name: ..., dependencies: [...]}, ...} :param circular: name of the service that is circularly dependent (used in recursion) :return: name of the service that is circularly dependent; None if there is no circular dependency ''' if circular is not None: return circular elif start in service_map.keys(): all_deps = service_map[start][var_names.dependencies] for dep in all_deps: if dep == check: return service_map[start][var_names.name] else: circ = service_circular_dependency(check, dep, service_map, circular) if circ is not None: return circ return None return None