diff options
Diffstat (limited to 'django/engagementmanager/service/authorization_service.py')
-rwxr-xr-x | django/engagementmanager/service/authorization_service.py | 605 |
1 files changed, 605 insertions, 0 deletions
diff --git a/django/engagementmanager/service/authorization_service.py b/django/engagementmanager/service/authorization_service.py new file mode 100755 index 0000000..c850b4a --- /dev/null +++ b/django/engagementmanager/service/authorization_service.py @@ -0,0 +1,605 @@ +# +# ============LICENSE_START========================================== +# org.onap.vvp/engagementmgr +# =================================================================== +# Copyright © 2017 AT&T Intellectual Property. All rights reserved. +# =================================================================== +# +# Unless otherwise specified, all software contained herein is licensed +# under the Apache License, Version 2.0 (the “License”); +# you may not use this software except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +# +# +# Unless otherwise specified, all documentation contained herein is licensed +# under the Creative Commons License, Attribution 4.0 Intl. (the “License”); +# you may not use this documentation except in compliance with the License. +# You may obtain a copy of the License at +# +# https://creativecommons.org/licenses/by/4.0/ +# +# Unless required by applicable law or agreed to in writing, documentation +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +# ============LICENSE_END============================================ +# +# ECOMP is a trademark and service mark of AT&T Intellectual Property. +import json +from enum import Enum +import rest_framework +from engagementmanager.models import Role, Engagement, Checklist, NextStep, VFC, \ + VF, ChecklistDecision, Notification +from engagementmanager.utils.constants import Roles +from engagementmanager.utils.request_data_mgr import request_data_mgr +from engagementmanager.service.logging_service import LoggingServiceFactory + +logger = LoggingServiceFactory.get_logger() + + +class Permissions(Enum): + """ + This ENUM holds all the actions that require permissions on the ICE portal + """ + update_user = 1, # TODO Any user can update their own account + add_vendor = 2, # only el or admin + add_nextstep = 3, # only el or admin + complete_nextstep = 4, # any eng member + delete_nextstep = 5, # only el or admin + approve_nextstep = 6, # only el or admin + deny_nextstep = 7, # only el or admin + reset_nextstep = 8 # any eng member + add_checklist = 9, # only el of a given engagement + set_checklist_decision = 10, # TODO only el defined as owner of cl + add_checklist_audit_log = 11, # TODO only el defined as owner of cl + delete_checklist_audit_log = 12, # Do we have this capability ?? + el_review_checklist = 13, # only checklist owner can set state. + # only el defined as peer_viewer of given engagement and is cl owner + peer_review_checklist = 14, + # TODO only admin which is defined as cl owner + admin_approve_checklist = 15, + # TODO only el of a given engagement and is the cl owner + handoff_checklist = 16, + add_vf = 17, # only standard_user + add_vfc = 18, # only standard_user ?? + delete_vfc = 19, # only standard_user + add_checklist_nextstep = 20, # only el + is_el_of_eng = 21, # only el of engagement + update_personal_next_step = 22, + create_checklist_audit_log = 23, + create_checklist_decision = 24, + update_checklist_decision = 25, + update_checklist_state = 26, + create_deployment_target_site = 27, + star_an_engagement = 28, + invite = 29, + update_account = 30, + set_ssh = 31, + update_password = 32, + get_engagement_status = 34, + put_engagement_status = 35, + eng_membership = 36, + delete_engagement = 37, + view_checklist = 38, # only non-standard-user + get_vfc = 39, + pull_activities = 40, + get_deployment_target_site = 41, + add_deployment_target_site = 42, + delete_deployment_target_site = 43, + export_engagments = 44, + archive_engagement = 45, + get_el_list = 46, + update_engagement = 47, + view_checklist_template = 48, + edit_checklist_template = 49, + delete_notification = 50, + remove_from_engagement_team = 51, + update_engagement_reviewers = 52, + edit_nextstep = 54, + order_nextstep = 60, + set_nextstep = 56, + edit_stage = 57, + edit_progress_bar = 58, + get_progress_bar = 59, + change_lab_entry = 62, + update_vf = 61, # only non-standard-user + add_feedback = 63, + + +class AuthorizationService: + """ + The Authorization Service detemines whether a given action is authorized for a specific user. + The method: is_user_able_to performs the authorization check given a user and an action (from Permissions ENUM) + """ + role_standard_user = None + role_el = None + role_admin = None + role_admin_ro = None + + def __get_role_checks(self, user, action): + """ + Returns the list of auth checks that should be performed per user action. + Returns None if the action is not listed in the authorized actions of the given user. + """ + result = None + + # EL # + if (user.role == self.role_el) and (action in self.el_permissions): + result = self.el_permissions[action] + # ADMIN # + elif user.role == self.role_admin and action in self.admin_permissions: + result = self.admin_permissions[action] + # ADMIN Read only # + elif user.role == self.role_admin_ro and action in self.admin_ro_permissions: + result = self.admin_ro_permissions[action] + # STANDRARD_USER # + if user.role == self.role_standard_user and action in self.standard_user_permissions: + result = self.standard_user_permissions[action] + + return result + + def __require_eng_membership(self, user, action, **kwargs): + """ + Determines whether a given user is part of engagement team by the eng uuid + user = IceUser + eng = UUID as a string + + :param user: user for auth check + :param action: action for auth check + :param kwargs: eng_uuid, checklist_uuid, ... + :return: Boolean, Message -> True/False if auth check succeeds/fails and a message describing auth failure + """ + eng = kwargs['eng'] + + try: + # @UndefinedVariable + if (user.email == eng.reviewer.email or user.email == eng.peer_reviewer.email or user.role.name == Roles.admin.name): + return True, 'OK' + else: + # validate if user in Team + if user in eng.engagement_team.all(): + return True, 'OK' + else: + return False, "" + except Engagement.DoesNotExist: + msg = 'User ' + user.email + ' is not a member of engagement: ' + eng.uuid + \ + ' / User is a not peer reviewer / admin of the engagement / Engagement wasnt found while fetching from DB' + logger.info(msg) + return False, msg + except Exception as e: + print(e) + msg = 'A general error occurred while trying to validate that User ' + \ + user.email + ' is a member of engagement ' + logger.info(msg + " Error: " + str(e)) + return False, msg + + def __require_peer_review_ownership(self, user, action, **kwargs): + """ + Determines whether the given user is the peer reviewer of the checklist + """ + + cl = kwargs['cl'] + eng = kwargs['eng'] + if cl and eng: + # @UndefinedVariable + if (eng.peer_reviewer == user and cl.owner == user) or (user.role.name == Roles.admin.name): + return True, 'OK' + else: + return False, 'User is either not the owner of the checklist or not a peer reviewer of the checklist' + else: + logger.error( + 'Internal Error - Checklist/Engagement not found while trying to check permissions for user ' + user.email) + return False, 'Internal Error - Checklist not found' + + def __require_cl_ownership(self, user, action, **kwargs): + """ + Determines whether the given user is the owner of the checklist + """ + + cl = kwargs['cl'] + if cl: + # @UndefinedVariable + if cl.owner == user or user.role.name == Roles.admin.name: + return True, 'OK' + else: + return False, 'User is not the owner of the checklist' + else: + logger.error( + 'Internal Error - Checklist not found while trying to check permissions for user ' + user.email) + return False, 'Internal Error - Checklist not found' + + def __require_el_of_engagement(self, user, action, **kwargs): + """ + Determines whether the given user is the el of the engagement + """ + eng = kwargs['eng'] + + if eng: + if (user.role.name == Roles.admin.name): # @UndefinedVariable + return True, 'OK' + if (user.uuid == eng.reviewer.uuid): # @UndefinedVariable + return True, 'OK' + + return False, 'Role Not authorized' + else: + logger.error( + 'Internal Error - Engagement not found while trying to check permissions for user ' + user.email) + return False, 'Internal Error - Checklist not found' + + def __noop(self, user, action, **kwargs): + """ + Do nothing, just authorize the action for the given user + """ + return True, 'OK' + + def __prevent(self, user, action, **kwargs): + """ + Do nothing, just prevent the action for the given user + """ + return False, 'Role Not authorized' + + def __is_notification_owner(self, user, action, **kwargs): + msg = 'Role Not authorized' + authorized = False + + notification_uuid = request_data_mgr.get_notification_uuid() + if notification_uuid: + if Notification.objects.get(uuid=notification_uuid).user == user: + authorized = True + msg = 'OK' + + return authorized, msg + + ###################### + # EL Permissions # + ###################### + """ + Each Permission Map is composed of the following key-val pairs: + Key='Action (Permission ENUM)' --> Value='Set of Checks to perform on this action.' + """ + el_permissions = { + Permissions.add_vf: {__noop}, + Permissions.add_feedback: {__noop}, + Permissions.update_user: {__noop}, + Permissions.add_vendor: {__noop}, + Permissions.update_vf: {__require_eng_membership}, + Permissions.add_nextstep: {__require_eng_membership}, + Permissions.complete_nextstep: {__require_eng_membership}, + Permissions.delete_nextstep: {__require_eng_membership}, + Permissions.order_nextstep: {__require_eng_membership}, + Permissions.set_nextstep: {__require_eng_membership}, + Permissions.edit_stage: {__require_eng_membership}, + Permissions.edit_progress_bar: {__require_eng_membership}, + Permissions.get_progress_bar: {__require_eng_membership}, + Permissions.change_lab_entry: {__require_eng_membership}, + Permissions.approve_nextstep: {__require_eng_membership}, + Permissions.deny_nextstep: {__require_eng_membership}, + Permissions.add_checklist: {__require_eng_membership}, + Permissions.set_checklist_decision: {__require_cl_ownership}, + Permissions.add_checklist_audit_log: {__require_cl_ownership}, + Permissions.delete_checklist_audit_log: {__require_cl_ownership}, + Permissions.el_review_checklist: {__require_cl_ownership, __require_eng_membership}, + Permissions.peer_review_checklist: {__require_peer_review_ownership}, + Permissions.handoff_checklist: {__require_cl_ownership, __require_eng_membership}, + Permissions.add_checklist_nextstep: {__require_cl_ownership, __require_eng_membership}, + Permissions.edit_nextstep: {__require_eng_membership}, + Permissions.is_el_of_eng: {__require_el_of_engagement}, + Permissions.update_personal_next_step: {__noop}, + Permissions.create_checklist_audit_log: {__require_eng_membership}, + Permissions.create_checklist_decision: {__require_eng_membership}, + Permissions.update_checklist_state: {__require_cl_ownership, __require_eng_membership}, + Permissions.create_deployment_target_site: {__require_eng_membership}, + Permissions.star_an_engagement: {__noop}, + Permissions.invite: {__require_eng_membership}, + Permissions.update_account: {__require_eng_membership}, + Permissions.set_ssh: {__require_eng_membership}, + Permissions.update_password: {}, + Permissions.delete_vfc: {__require_eng_membership}, + Permissions.get_engagement_status: {__require_eng_membership}, + Permissions.put_engagement_status: {__require_eng_membership}, + Permissions.eng_membership: {__noop}, + Permissions.delete_engagement: {__require_eng_membership}, + Permissions.view_checklist: {__require_eng_membership}, + Permissions.pull_activities: {__require_eng_membership}, + Permissions.get_deployment_target_site: {__noop}, + Permissions.add_deployment_target_site: {__noop}, + Permissions.delete_deployment_target_site: {__noop}, + Permissions.export_engagments: {__noop}, + Permissions.update_checklist_decision: {__noop}, + Permissions.get_vfc: {__noop}, + Permissions.add_vfc: {__noop}, + Permissions.delete_notification: {__is_notification_owner}, + Permissions.update_engagement: {__noop}, + Permissions.remove_from_engagement_team: {__require_eng_membership}, + } + + ################################# + # STANDARD_USER Permissions # + ################################# + standard_user_permissions = { + Permissions.update_user: {__noop}, + Permissions.add_vf: {__noop}, + Permissions.add_feedback: {__noop}, + Permissions.add_vfc: {__noop}, + Permissions.get_vfc: {__require_eng_membership}, + Permissions.delete_vfc: {__require_eng_membership}, + Permissions.complete_nextstep: {__require_eng_membership}, + Permissions.update_vf: {__require_eng_membership}, + Permissions.reset_nextstep: {__require_eng_membership}, + Permissions.update_personal_next_step: {__noop}, + Permissions.update_checklist_state: {__require_cl_ownership, __require_eng_membership}, + Permissions.create_deployment_target_site: {__require_eng_membership}, + Permissions.star_an_engagement: {__noop}, + Permissions.invite: {__require_eng_membership}, + Permissions.update_account: {__require_eng_membership}, + Permissions.set_ssh: {__require_eng_membership}, + Permissions.update_password: {__require_eng_membership}, + Permissions.delete_vfc: {__require_eng_membership}, + Permissions.get_engagement_status: {__require_eng_membership}, + Permissions.eng_membership: {__noop}, + Permissions.pull_activities: {__require_eng_membership}, + Permissions.export_engagments: {__noop}, + Permissions.update_checklist_decision: {__noop}, + Permissions.remove_from_engagement_team: {__require_eng_membership}, + Permissions.delete_notification: {__is_notification_owner}, + Permissions.change_lab_entry: {__require_eng_membership}, + } + + ###################### + # ADMIN Permissions # + ###################### + ###################################################### + # TODO: We need to decide exactly what are the ADMIN + # TODO: permissions. Currently it matches EL + + # TODO: admin_approve_checklist + ###################################################### + admin_permissions = dict(el_permissions) # Duplicate permissions of EL + admin_permissions.update( # Add Extra permissions to admin + { + Permissions.admin_approve_checklist: {__require_cl_ownership}, + Permissions.remove_from_engagement_team: {__require_eng_membership}, + Permissions.view_checklist_template: {__noop}, + Permissions.edit_checklist_template: {__noop}, + Permissions.archive_engagement: {__noop}, + Permissions.get_el_list: {__noop}, + Permissions.update_engagement_reviewers: {__noop}, + Permissions.edit_nextstep: {__noop}, + Permissions.delete_nextstep: {__noop}, + Permissions.order_nextstep: {__noop}, + Permissions.set_nextstep: {__noop}, + Permissions.edit_stage: {__noop}, + Permissions.edit_progress_bar: {__noop}, + Permissions.get_progress_bar: {__noop}, + Permissions.change_lab_entry: {__noop}, + } + ) + + ###################### + # ADMIN Read only Permissions # + ###################### + admin_ro_permissions = dict() + admin_ro_permissions.update( # Add Extra permissions to admin_ro + { + Permissions.add_vf: {__prevent}, + Permissions.add_feedback: {__noop}, + Permissions.get_vfc: {__noop}, + Permissions.get_engagement_status: {__noop}, + Permissions.eng_membership: {__noop}, + Permissions.pull_activities: {__noop}, + Permissions.star_an_engagement: {__noop}, + Permissions.export_engagments: {__noop}, + } + ) + + def __init__(self): + self.role_standard_user = self.role_el = self.role_admin = self.role_admin_ro = None + self.__load_roles_from_db() + + def check_permissions(self, user, action, eng_uuid, role, eng, cl): + # Retrieve the permission checks that should be performed on this user + # role and action + perm_checks = self.__get_role_checks(user, action) + if not perm_checks: + # Permission Checks were not found, it means that the action is not listed in the permitted + # actions for the role of the user + ret = False, 'Role ' + str(role.name) + ' is not permitted to ' + \ + str(action.name) + '/ Engagement: ' + \ + str(eng_uuid) + " isn't valid" + else: + # Start invoking permissions checks one by one. + for check in perm_checks: + ret = result, message = check( + self, user, action, eng=eng, cl=cl) + if result: + # Permission check succeeded + continue + else: + break # Permission check failed + + return ret + + """ + Determines whether a user is able to perform some action. + """ + + def is_user_able_to(self, user, action, eng_uuid, checklist_uuid): + role = user.role + ret = True, 'OK' + + checklist_uuid = request_data_mgr.get_cl_uuid() + + # Retrieve Engagement and Checklist if their UUIDs were supplied + eng, cl = self.__get_objects_from_db(eng_uuid, checklist_uuid) + if eng and not eng_uuid: + eng_uuid = eng.uuid + + ret = self.check_permissions(user, action, eng_uuid, role, eng, cl) + + return ret + + def __get_objects_from_db(self, eng_uuid, cl_uuid): + eng = cl = None + + try: + if eng_uuid: + eng = Engagement.objects.get(uuid=eng_uuid) + except Engagement.DoesNotExist: + logger.error( + 'ENG was not found while checking permissions... returning 500') + return None, None + + try: + if cl_uuid: + cl = Checklist.objects.get(uuid=cl_uuid) + if not eng: + eng = cl.engagement + except Checklist.DoesNotExist: + logger.error('CL was not found while checking permissions') + cl = None + + return eng, cl + + def __load_roles_from_db(self): + self.role_standard_user, created = Role.objects.get_or_create( + name=Roles.standard_user.name) # @UndefinedVariable + self.role_el, created = Role.objects.get_or_create( + name=Roles.el.name) # @UndefinedVariable + self.role_admin, created = Role.objects.get_or_create( + name=Roles.admin.name) # @UndefinedVariable + self.role_admin_ro, created = Role.objects.get_or_create( + name=Roles.admin_ro.name) # @UndefinedVariable + + def prepare_data_for_auth(self, *args, **kwargs): + eng_uuid = None + # Extract ENG_UUID # + if 'eng_uuid' in kwargs: + eng_uuid = kwargs['eng_uuid'] + elif 'engagement_uuid' in kwargs: + eng_uuid = kwargs['engagement_uuid'] + else: + # Extract eng_uuid from request body + for arg in args: + if eng_uuid != None: + break + if isinstance(arg, rest_framework.request.Request): + try: + if arg.body: + data = json.loads(arg.body) + try: + iter(data) + for item in data: + if 'eng_uuid' in item and item['eng_uuid']: + eng_uuid = item['eng_uuid'] + break + elif 'eng_uuid' in item and item.eng_uuid: + eng_uuid = item.eng_uuid + break + elif item == 'eng_uuid': + eng_uuid = item + break + except TypeError: + if 'eng_uuid' in data and data['eng_uuid']: + eng_uuid = data['eng_uuid'] + + elif 'engagement_uuid' in data and data['engagement_uuid']: + eng_uuid = data['engagement_uuid'] + except Exception as e: + print(e) + pass + + request_data_mgr.set_eng_uuid(eng_uuid) + + # Extract CHECKLIST_UUID # + if 'checklistUuid' in kwargs: + request_data_mgr.set_cl_uuid(kwargs['checklistUuid']) + if (eng_uuid == None): + try: + eng_uuid = Checklist.objects.get( + uuid=request_data_mgr.get_cl_uuid()).engagement.uuid + request_data_mgr.set_eng_uuid(eng_uuid) + except Checklist.DoesNotExist: + raise Exception("auth service couldn't fetch Checklist by checklist uuid=" + + request_data_mgr.get_cl_uuid()) + except Exception as e: + raise Exception( + "Failed fetching engagement uuid from checklist " + request_data_mgr.get_cl_uuid()) + + # Extract engagement by NEXTSTEP_UUID # + if 'ns_uuid' in kwargs: + request_data_mgr.set_ns_uuid(kwargs['ns_uuid']) + if (eng_uuid == None): + next_step = None + try: + next_step = NextStep.objects.get( + uuid=request_data_mgr.get_ns_uuid()) + except NextStep.DoesNotExist: + raise Exception("auth service couldn't fetch NextStep by nextstep uuid=" + + request_data_mgr.get_ns_uuid()) + + try: + eng_uuid = next_step.engagement.uuid + request_data_mgr.set_eng_uuid(eng_uuid) + except: + # If we've gotten here it means that the next_step doesn't have attached + # engagement (e.g personal next_step) + pass + + # Extract engagement by VFC + if ('uuid' in kwargs): + from engagementmanager.rest.vfc import VFCRest + if (isinstance(args[0], VFCRest) == True): + try: + vfc = VFC.objects.get(uuid=kwargs['uuid']) + if (eng_uuid == None): + eng_uuid = vfc.vf.engagement.uuid + request_data_mgr.set_eng_uuid(eng_uuid) + except VFC.DoesNotExist: + raise Exception( + "auth service couldn't fetch vfc by vfc uuid=" + kwargs['uuid']) + + # Extract engagement by VF (unfortunately the url exposed by the server + # get uuid as a parameter and serve both vf and vfc APIs) # + if 'vf_uuid' in kwargs and eng_uuid == None: + try: + eng_uuid = VF.objects.get( + uuid=kwargs['vf_uuid']).engagement.uuid + request_data_mgr.set_eng_uuid(eng_uuid) + except VF.DoesNotExist: + logger.error( + "Prepare_data_for_auth: Couldn't fetch engagement object from VF, trying to fetch from VFC...") + vfc = None + try: + vfc = VFC.objects.get(uuid=kwargs['vf_uuid']) + if (vfc != None): + eng_uuid = vfc.vf.engagement.uuid + request_data_mgr.set_eng_uuid(eng_uuid) + except VFC.DoesNotExist: + logger.error( + "Prepare_data_for_auth: Couldn't fetch engagement object from VFC") + + # Extract engagement by ChecklistDecision + if 'decision_uuid' in kwargs and eng_uuid == None: + try: + eng_uuid = ChecklistDecision.objects.get( + uuid=kwargs['decision_uuid']).checklist.engagement.uuid + request_data_mgr.set_eng_uuid(eng_uuid) + except ChecklistDecision.DoesNotExist: + logger.error( + "Prepare_data_for_auth: Couldn't fetch engagement object from ChecklistDecision") + + # Extract notification uuid for permission check + if 'notif_uuid' in kwargs: + request_data_mgr.set_notification_uuid(kwargs['notif_uuid']) + + return eng_uuid |