"""Validate state params.""" from http import HTTPStatus from flask import abort from abacus_state.constants.constants import ( DOCUMENTS_PROFILE_ALLOWED_ACTIONS_BY_PARENT_TABLE as allowed_actions, DOCUMENTS_PROFILE_ALLOWED_STATUSES_BY_PARENT_TABLE_BY_ACTION_NAMES as allowed_statuses, ORCHARD_PROFILE_TYPES, ) from abacus_state.constants.error import ( ERROR_DONT_HAVE_PERMISSIONS, ERROR_MESSAGE_FORBIDDEN_PROFILE_TYPE_CREATE, ERROR_MESSAGE_FORBIDDEN_PROFILE_TYPE_UPDATE, ) from abacus_state.models.abacus_state import AbacusState from abacus_state.utils.validation_access_checks import ( build_resource_access_check, validate_record_owner_access_checks, ) def check_create_permissions_access_checks( params: list[dict[str, str]], profile_type: str | None ): """Validate profile type for to have permission on creating states. Args: params: action states profile_type: profile type, which should come from header """ if profile_type != ORCHARD_PROFILE_TYPES.DOCUMENTS_PROFILE: return access_checks = [] for item in params: parent_table_name = item['parent_table_name'] parent_table_id = item['parent_table_id'] action = item['action_name'] if ( parent_table_name not in allowed_actions or action not in allowed_actions[parent_table_name] ): abort( code=HTTPStatus.BAD_REQUEST, description=ERROR_MESSAGE_FORBIDDEN_PROFILE_TYPE_CREATE.format( profile_type=profile_type, action=action, table=parent_table_name ), ) access_check = build_resource_access_check(parent_table_name, parent_table_id) access_checks.append(access_check) if not validate_record_owner_access_checks(access_checks): abort(code=HTTPStatus.UNAUTHORIZED, description=ERROR_DONT_HAVE_PERMISSIONS) def check_update_permissions_access_checks( obj: AbacusState, params: dict[str, str], profile_type: str | None ): """Validate profile type to have permission on updating states. Args: obj: state object params: update params profile_type: profile type, which should come from header """ if profile_type != ORCHARD_PROFILE_TYPES.DOCUMENTS_PROFILE: return parent_table_name = obj.parent_table_name parent_table_id = obj.parent_table_id action = obj.action_name action_status = params.get('action_status') if ( parent_table_name not in allowed_actions or action not in allowed_actions[parent_table_name] or action_status not in allowed_statuses.get((parent_table_name, action), []) ): abort( code=HTTPStatus.BAD_REQUEST, description=ERROR_MESSAGE_FORBIDDEN_PROFILE_TYPE_UPDATE.format( profile_type=profile_type, action=action, table=parent_table_name, action_status=action_status, ), ) access_check = build_resource_access_check(parent_table_name, parent_table_id) if not validate_record_owner_access_checks([access_check]): abort(code=HTTPStatus.UNAUTHORIZED, description=ERROR_DONT_HAVE_PERMISSIONS)