lavkach3

1
import os
2
from enum import Enum
3
from uuid import UUID
4

5
import yaml
6
from sqlalchemy import select
7
from starlette.exceptions import HTTPException
8

9
#from app.basic.user.models import Role
10
#from core.service.base import BaseService
11

12
Allow = 'ALLOW'
13
Deny = 'Deny'
14

15

16
class Permissions(Enum):
17
    ...
18

19

20
def get_all_permit_fiels():
21
    paths = []
22
    path = os.path.abspath(__file__).split('/')
23
    ''.join(path[:-3])
24
    for root, dirs, files in os.walk('/'.join(path[:-3]) + '/app'):
25
        for file in files:
26
            if file.endswith(".yaml"):
27
                paths.append(os.path.join(root, file))
28
    return paths
29

30

31
paths = get_all_permit_fiels()
32
permits = {}
33
for path in paths:
34
    with open(path, "r") as stream:
35
        try:
36
            yam = yaml.safe_load(stream)
37
            if yam:
38
                for k, v in yam.get('permits').items():
39
                    permits.update(
40
                        {k: v}
41
                    )
42
        except yaml.YAMLError as exc:
43
            print(exc)
44

45

46
def permit(*arg):
47
    """
48
    На вход дается пермит, далее логика такова
49
    Ищется в списке по пермиту роли, если роль сразу есть в списке пользователя, то функция дает True и заканчивается
50
    Если же роль не найдена, то функия идет рекурсивно по родителям ролей и так же пытается найти роль пользователя в родителях
51
    """
52

53
    def inner_decorator(f):
54
        async def wrapped(*args, **kwargs):
55

56
            service = None
57
            res = False
58
            for a in args:
59
                if isinstance(a, object):
60
                    service = a
61
                    break
62
            if not service:
63
                raise HTTPException(status_code=403, detail=f"User not found")
64
            role = service.env['role'].model
65
            if not service.user.is_admin:
66
                service_roles = [UUID(i) for i in service.user.role_ids]
67
                query = select(role).where(
68
                    role.permission_allow_list.contains(arg)
69
                ).where(
70
                    role.company_id.in_(service.user.company_ids)
71
                )
72
                result = await service.session.execute(query)
73
                roles = result.scalars().all()
74

75
                if set(service_roles) & set([i.id for i in roles]):
76
                    res = True
77
                if not res:
78
                    parents = []
79
                    for r in roles:
80
                        if r.role_ids:
81
                            parents += r.role_ids
82
                    while parents:
83
                        query = select(role).where(
84
                            role.id.in_(parents)
85
                        ).where(
86
                            role.company_id.in_(service.user.company_ids)
87
                        )
88
                        result = await service.session.execute(query)
89
                        roles = result.scalars().all()
90
                        if set(service_roles) & set([i.id for i in roles]):
91
                            res = True
92
                            break
93
                        parents = []
94
                        for r in roles:
95
                            if r.parents:
96
                                parents += r.parents
97
                if not res:
98
                    raise HTTPException(status_code=403,
99
                                        detail=f"The user ({service.user}) does not have permission to {arg}")
100
            response = f(*args, **kwargs)
101
            return await response
102

103
        return wrapped
104

105
    return inner_decorator
106