# # ============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