-
Notifications
You must be signed in to change notification settings - Fork 49
/
models.py
273 lines (218 loc) · 8.17 KB
/
models.py
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
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
# -*- coding: utf-8 -*-
#
# This file is part of Invenio.
# Copyright (C) 2015-2023 CERN.
#
# Invenio is free software; you can redistribute it and/or modify it
# under the terms of the MIT License; see LICENSE file for more details.
"""Database models for access module."""
from flask_principal import RoleNeed, UserNeed
from invenio_accounts.models import Role, User
from invenio_db import db
from sqlalchemy import UniqueConstraint
from sqlalchemy.event import listen
from sqlalchemy.orm import validates
from sqlalchemy.orm.attributes import get_history
from .proxies import current_access
class ActionNeedMixin(object):
"""Define common attributes for Action needs."""
id = db.Column(db.Integer, autoincrement=True, primary_key=True)
"""Primary key. It allows the other fields to be nullable."""
action = db.Column(db.String(80), index=True)
"""Name of the action."""
exclude = db.Column(
db.Boolean(name="exclude"), nullable=False, default=False, server_default="0"
)
"""If set to True, deny the action, otherwise allow it."""
argument = db.Column(db.String(255), nullable=True, index=True)
"""Action argument."""
@classmethod
def create(cls, action, **kwargs):
"""Create new database row using the provided action need.
:param action: An object containing a method equal to ``'action'`` and
a value.
:param argument: The action argument. If this parameter is not passed,
then the ``action.argument`` will be used instead. If the
``action.argument`` does not exist, ``None`` will be set as
argument for the new action need.
:returns: An :class:`invenio_access.models.ActionNeedMixin` instance.
"""
assert action.method == "action"
argument = kwargs.pop("argument", None) or getattr(action, "argument", None)
return cls(action=action.value, argument=argument, **kwargs)
@classmethod
def allow(cls, action, **kwargs):
"""Allow the given action need.
:param action: The action to allow.
:returns: A :class:`invenio_access.models.ActionNeedMixin` instance.
"""
return cls.create(action, exclude=False, **kwargs)
@classmethod
def deny(cls, action, **kwargs):
"""Deny the given action need.
:param action: The action to deny.
:returns: A :class:`invenio_access.models.ActionNeedMixin` instance.
"""
return cls.create(action, exclude=True, **kwargs)
@classmethod
def query_by_action(cls, action, argument=None):
"""Prepare query object with filtered action.
:param action: The action to deny.
:param argument: The action argument. If it's ``None`` then, if exists,
the ``action.argument`` will be taken. In the worst case will be
set as ``None``. (Default: ``None``)
:returns: A query object.
"""
query = cls.query.filter_by(action=action.value)
argument = argument or getattr(action, "argument", None)
if argument is not None:
query = query.filter(
db.or_(
cls.argument == str(argument),
cls.argument.is_(None),
)
)
else:
query = query.filter(cls.argument.is_(None))
return query
@property
def need(self):
"""Return the need corresponding to this model instance.
This is an abstract method and will raise NotImplementedError.
"""
raise NotImplementedError() # pragma: no cover
class ActionUsers(ActionNeedMixin, db.Model):
"""ActionUsers data model.
It relates an allowed action with a user.
"""
__tablename__ = "access_actionsusers"
__table_args__ = (
UniqueConstraint(
"action",
"exclude",
"argument",
"user_id",
name="access_actionsusers_unique",
),
)
user_id = db.Column(
db.Integer(),
db.ForeignKey(User.id, ondelete="CASCADE"),
nullable=False,
index=True,
)
user = db.relationship(
"User", backref=db.backref("actionusers", cascade="all, delete-orphan")
)
@property
def need(self):
"""Return UserNeed instance."""
return UserNeed(self.user_id)
class ActionRoles(ActionNeedMixin, db.Model):
"""ActionRoles data model.
It relates an allowed action with a role.
"""
__tablename__ = "access_actionsroles"
__table_args__ = (
UniqueConstraint(
"action",
"exclude",
"argument",
"role_id",
name="access_actionsroles_unique",
),
)
role_id = db.Column(
db.String(80),
db.ForeignKey(Role.id, ondelete="CASCADE"),
nullable=False,
index=True,
)
role = db.relationship(
"Role", backref=db.backref("actionusers", cascade="all, delete-orphan")
)
@property
def need(self):
"""Return RoleNeed instance."""
return RoleNeed(self.role.name)
class ActionSystemRoles(ActionNeedMixin, db.Model):
"""ActionSystemRoles data model.
It relates an allowed action with a predefined role.
Example: "any user"
"""
__tablename__ = "access_actionssystemroles"
__table_args__ = (
UniqueConstraint(
"action",
"exclude",
"argument",
"role_name",
name="access_actionssystemroles_unique",
),
)
role_name = db.Column(db.String(40), nullable=False, index=True)
@classmethod
def create(cls, action, **kwargs):
"""Create new database row using the provided action need."""
role = kwargs.pop("role", None)
if role:
assert role.method == "system_role"
kwargs["role_name"] = role.value
return super(ActionSystemRoles, cls).create(action, **kwargs)
@validates("role_name")
def validate_role_name(self, key, role_name):
"""Checks that the role name has been registered."""
assert role_name in current_access.system_roles
return role_name
@property
def need(self):
"""Return the corresponding Need instance."""
return current_access.system_roles[self.role_name]
def get_action_cache_key(name, argument):
"""Get an action cache key string."""
tokens = [str(name)]
if argument:
tokens.append(str(argument))
return "::".join(tokens)
def removed_or_inserted_action(mapper, connection, target):
"""Remove the action from cache when an item is inserted or deleted."""
current_access.delete_action_cache(
get_action_cache_key(target.action, target.argument)
)
def changed_action(mapper, connection, target):
"""Remove the action from cache when an item is updated."""
action_history = get_history(target, "action")
argument_history = get_history(target, "argument")
owner_history = get_history(
target,
"user"
if isinstance(target, ActionUsers)
else "role"
if isinstance(target, ActionRoles)
else "role_name",
)
if (
action_history.has_changes()
or argument_history.has_changes()
or owner_history.has_changes()
):
current_access.delete_action_cache(
get_action_cache_key(target.action, target.argument)
)
current_access.delete_action_cache(
get_action_cache_key(
action_history.deleted[0] if action_history.deleted else target.action,
argument_history.deleted[0]
if argument_history.deleted
else target.argument,
)
)
listen(ActionUsers, "after_insert", removed_or_inserted_action)
listen(ActionUsers, "after_delete", removed_or_inserted_action)
listen(ActionUsers, "after_update", changed_action)
listen(ActionRoles, "after_insert", removed_or_inserted_action)
listen(ActionRoles, "after_delete", removed_or_inserted_action)
listen(ActionRoles, "after_update", changed_action)
listen(ActionSystemRoles, "after_insert", removed_or_inserted_action)
listen(ActionSystemRoles, "after_delete", removed_or_inserted_action)
listen(ActionSystemRoles, "after_update", changed_action)