-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathauthorization.py
More file actions
171 lines (134 loc) · 5.11 KB
/
Copy pathauthorization.py
File metadata and controls
171 lines (134 loc) · 5.11 KB
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
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
from dataclasses import dataclass
from typing import Protocol
from src.modules.authorization import AuthorizationResource, Permission, Role
from src.modules.authorization.domain.permissions import (
DEFAULT_POLICIES,
DEFAULT_RESOURCES,
DEFAULT_ROLES,
)
class AuthorizationSeedRepository(Protocol):
async def list_resources(self) -> list[AuthorizationResource]:
raise NotImplementedError
async def create_resource(
self,
resource: AuthorizationResource,
) -> AuthorizationResource:
raise NotImplementedError
async def list_roles(self) -> list[Role]:
raise NotImplementedError
async def create_role(self, role: Role) -> Role:
raise NotImplementedError
async def list_permissions(self) -> list[Permission]:
raise NotImplementedError
async def create_permission(self, permission: Permission) -> Permission:
raise NotImplementedError
async def assign_permission_to_role(
self,
role_id,
permission_id,
) -> None:
raise NotImplementedError
async def add_policy(self, ptype: str, *values: str) -> None:
raise NotImplementedError
@dataclass(frozen=True)
class AuthorizationSeedResult:
resources_created: int = 0
roles_created: int = 0
permissions_created: int = 0
role_permissions_created: int = 0
policies_created: int = 0
async def seed_authorization(
repository: AuthorizationSeedRepository,
) -> AuthorizationSeedResult:
existing_resources = {
resource.key: resource for resource in await repository.list_resources()
}
existing_roles = {role.name: role for role in await repository.list_roles()}
existing_permissions = {
permission.key: permission for permission in await repository.list_permissions()
}
existing_role_permissions = await _load_role_permissions(repository)
existing_policies = await _load_policies(repository)
resources_created = 0
for resource_definition in DEFAULT_RESOURCES:
if resource_definition.key in existing_resources:
continue
resource = await repository.create_resource(
AuthorizationResource.create(
key=resource_definition.key,
name=resource_definition.name,
description=resource_definition.description,
)
)
existing_resources[resource.key] = resource
resources_created += 1
roles_created = 0
for name, description in _default_roles().items():
if name in existing_roles:
continue
role = await repository.create_role(
Role.create(name=name, description=description)
)
existing_roles[role.name] = role
roles_created += 1
permissions_created = 0
for key in _default_permission_keys():
if key in existing_permissions:
continue
resource, action = key.split(":", 1)
permission = await repository.create_permission(
Permission.create(
key=key,
resource=resource,
action=action,
description=f"Allows {action} access on {resource}",
)
)
existing_permissions[permission.key] = permission
permissions_created += 1
role_permissions_created = 0
for _, role_name, permission_key in _permission_policies():
role_permission = (role_name, permission_key)
if role_permission in existing_role_permissions:
continue
await repository.assign_permission_to_role(
existing_roles[role_name].id,
existing_permissions[permission_key].id,
)
existing_role_permissions.add(role_permission)
role_permissions_created += 1
policies_created = 0
for policy in DEFAULT_POLICIES:
if policy in existing_policies:
continue
ptype, *values = policy
await repository.add_policy(ptype, *values)
existing_policies.add(policy)
policies_created += 1
return AuthorizationSeedResult(
resources_created=resources_created,
roles_created=roles_created,
permissions_created=permissions_created,
role_permissions_created=role_permissions_created,
policies_created=policies_created,
)
def _default_roles() -> dict[str, str]:
return {role.name: role.description for role in DEFAULT_ROLES}
def _default_permission_keys() -> list[str]:
return [permission_key for _, _, permission_key in _permission_policies()]
def _permission_policies() -> list[tuple[str, str, str]]:
return [
policy for policy in DEFAULT_POLICIES if policy[0] == "p" and policy[2] != "*"
]
async def _load_role_permissions(
repository: AuthorizationSeedRepository,
) -> set[tuple[str, str]]:
if not hasattr(repository, "list_role_permissions"):
return set()
return set(await repository.list_role_permissions())
async def _load_policies(
repository: AuthorizationSeedRepository,
) -> set[tuple[str, ...]]:
if not hasattr(repository, "list_policies"):
return set()
return set(await repository.list_policies())