hydroserver-api-services/domains/iam/models/utils.py at d1e4a280fb48f8cc30d702d8fea48a4df2eb8e00 · hydroserver2/hydroserver-api-services · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
import typing
from typing import Optional, Union
from .permission import Permission

if typing.TYPE_CHECKING:
    from domains.iam.models import Workspace, APIKey
    from django.contrib.auth import get_user_model

    User = get_user_model()


class PermissionChecker:
    @classmethod
    def check_create_permissions(
        cls,
        principal: Optional[Union["User", "APIKey"]],
        workspace: Optional["Workspace"],
        resource_type: str,
    ):
        if not principal:
            return False

        if hasattr(principal, "account_type"):
            if principal.account_type in [
                "admin",
                "staff",
            ]:
                if not workspace and resource_type not in [
                    "OrchestrationSystem",
                    "DataConnection",
                    "ProcessingLevel",
                    "Unit",
                    "Sensor",
                    "ResultQualifier",
                    "ObservedProperty",
                ]:
                    return False
                return True

            if not workspace:
                return False

            if workspace.owner == principal:
                return True

            permissions = Permission.objects.filter(
                role__collaborator_assignments__user=principal,
                role__collaborator_assignments__workspace=workspace,
                resource_type__in=["*", resource_type],
            ).values_list("permission_type", flat=True)

        elif hasattr(principal, "workspace"):
            if not workspace or principal.workspace != workspace:
                return False

            permissions = Permission.objects.filter(
                role=principal.role,
                resource_type__in=["*", resource_type],
            ).values_list("permission_type", flat=True)

        else:
            return False

        return any(perm in permissions for perm in ["*", "create"])

    @staticmethod
    def check_object_permissions(
        principal: Optional[Union["User", "APIKey"]],
        workspace: Optional["Workspace"],
        resource_type: str,
    ):
        if not workspace:
            if hasattr(principal, "account_type") and principal.account_type in [
                "admin",
                "staff",
            ]:
                return ["view", "edit", "delete"]
            else:
                return ["view"]

        if hasattr(principal, "account_type"):
            if workspace.owner == principal or principal.account_type in [
                "admin",
                "staff",
            ]:
                return ["view", "edit", "delete"]

            permissions = list(
                Permission.objects.exclude(permission_type="create")
                .filter(
                    role__collaborator_assignments__user=principal,
                    role__collaborator_assignments__workspace=workspace,
                    resource_type__in=["*", resource_type],
                )
                .values_list("permission_type", flat=True)
            )

        elif hasattr(principal, "workspace"):
            if principal.workspace != workspace:
                permissions = []
            else:
                permissions = list(
                    Permission.objects.exclude(permission_type="create")
                    .filter(
                        role=principal.role,
                        resource_type__in=["*", resource_type],
                    )
                    .values_list("permission_type", flat=True)
                )

        else:
            permissions = []

        if "*" in permissions:
            permissions = ["view", "edit", "delete"]

        if not workspace.is_private and "view" not in permissions:
            if resource_type not in [
                "Thing",
                "Datastream",
                "OrchestrationSystem",
                "DataConnection",
                "Task",
                "APIKey",
            ]:
                permissions.append("view")

        return permissions