| 33 | |
| 34 | |
| 35 | class PolicyEngine: |
| 36 | session: Session |
| 37 | |
| 38 | def __init__(self, session: Session, project: 'Project'): |
| 39 | self.session = session |
| 40 | self.project = project |
| 41 | |
| 42 | def get_policy_enforcer(self, object_type: Enum) -> 'BasePolicyEnforcer': |
| 43 | from shared.permissions.policy_engine.base_policy_enforcer import BasePolicyEnforcer |
| 44 | from shared.permissions.policy_engine.dataset_policy_enforcer import DatasetPolicyEnforcer |
| 45 | from shared.permissions.policy_engine.file_policy_enforcer import FilePolicyEnforcer |
| 46 | POLICY_ENFORCERS_MAPPERS = { |
| 47 | 'WorkingDir': DatasetPolicyEnforcer, |
| 48 | 'File': FilePolicyEnforcer |
| 49 | } |
| 50 | enforcer_class = POLICY_ENFORCERS_MAPPERS.get(object_type.name) |
| 51 | if enforcer_class is None: |
| 52 | enforcer_class = BasePolicyEnforcer |
| 53 | return enforcer_class |
| 54 | |
| 55 | def member_has_perm(self, |
| 56 | member: Member, |
| 57 | object_type: Enum, |
| 58 | object_id: int, |
| 59 | perm: Enum) -> PermissionResult: |
| 60 | if member.user and member.user.is_super_admin: |
| 61 | result = PermissionResult( |
| 62 | allowed = True, |
| 63 | member_id = member.id, |
| 64 | object_type = object_type.name, |
| 65 | object_id = object_id |
| 66 | ) |
| 67 | return result |
| 68 | PolicyEnforcer = self.get_policy_enforcer(object_type = object_type) |
| 69 | enforcer = PolicyEnforcer(session = self.session, project = self.project, policy_engine = self) |
| 70 | perm_result = enforcer.has_perm(member_id = member.id, |
| 71 | object_type = object_type.name, |
| 72 | object_id = object_id, |
| 73 | perm = perm) |
| 74 | return perm_result |
| 75 | |
| 76 | def member_has_any_project_role(self, |
| 77 | member: Member, |
| 78 | roles: list, |
| 79 | project_id: int) -> PermissionResult: |
| 80 | from shared.database.permissions.roles import Role, RoleMemberObject, ValidObjectTypes |
| 81 | if not roles: |
| 82 | return False |
| 83 | role_member_objects = self.session.query(RoleMemberObject).join(Role, |
| 84 | Role.id == RoleMemberObject.role_id).filter( |
| 85 | RoleMemberObject.object_type == ValidObjectTypes.project.name, |
| 86 | Role.name.in_(roles), |
| 87 | RoleMemberObject.object_id == project_id, |
| 88 | RoleMemberObject.member_id == member.id |
| 89 | ) |
| 90 | |
| 91 | allowed = role_member_objects.first() is not None |
| 92 | if member.user and member.user.is_super_admin: |
no outgoing calls