[b6dba6]: Allura / allura / lib / security.py  Maximize  Restore  History

Download this file

336 lines (291 with data), 12.2 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
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
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
"""
This module provides the security predicates used in decorating various models.
"""
import logging
from collections import defaultdict
from pylons import c, request
from webob import exc
from itertools import chain
from ming.utils import LazyProperty
from allura.lib.utils import TruthyCallable
log = logging.getLogger(__name__)
class Credentials(object):
'''
Role graph logic & caching
'''
def __init__(self):
self.clear()
@classmethod
def get(cls):
'get the global Credentials instance'
import allura
return allura.credentials
def clear(self):
'clear cache'
self.users = {}
self.projects = {}
def load_user_roles(self, user_id, *project_ids):
'''Load the credentials with all user roles for a set of projects'''
from allura import model as M
# Don't reload roles
project_ids = [ pid for pid in project_ids if self.users.get((user_id, pid)) is None ]
if not project_ids: return
if user_id is None:
q = M.ProjectRole.query.find(
dict(
project_id={'$in': project_ids},
name='*anonymous'))
else:
q0 = M.ProjectRole.query.find(
dict(project_id={'$in': list(project_ids)},
name={'$in':['*anonymous', '*authenticated']}))
q1 = M.ProjectRole.query.find(
dict(project_id={'$in': list(project_ids)},user_id=user_id))
q = chain(q0, q1)
roles_by_project = dict((pid, []) for pid in project_ids)
for role in q:
roles_by_project[role.project_id].append(role)
for pid, roles in roles_by_project.iteritems():
self.users[user_id, pid] = RoleCache(self, roles)
def load_project_roles(self, *project_ids):
'''Load the credentials with all user roles for a set of projects'''
from allura import model as M
# Don't reload roles
project_ids = [ pid for pid in project_ids if self.projects.get(pid) is None ]
if not project_ids: return
q = M.ProjectRole.query.find(dict(
project_id={'$in': project_ids}))
roles_by_project = dict((pid, []) for pid in project_ids)
for role in q:
roles_by_project[role.project_id].append(role)
for pid, roles in roles_by_project.iteritems():
self.projects[pid] = RoleCache(self, roles)
def project_roles(self, project_id):
'''
:returns: a RoleCache of ProjectRoles for project_id
'''
roles = self.projects.get(project_id)
if roles is None:
self.load_project_roles(project_id)
roles = self.projects[project_id]
return roles
def user_roles(self, user_id, project_id=None):
'''
:returns: a RoleCache of ProjectRoles for given user_id and project_id, *anonymous and *authenticated checked as appropriate
'''
from allura import model as M
roles = self.users.get((user_id, project_id))
if roles is None:
if project_id is None:
if user_id is None:
q = []
else:
q = M.ProjectRole.query.find(dict(user_id=user_id))
roles = RoleCache(self, q)
else:
self.load_user_roles(user_id, project_id)
roles = self.users.get((user_id, project_id))
self.users[user_id, project_id] = roles
return roles
def user_has_any_role(self, user_id, project_id, role_ids):
user_roles = self.user_roles(user_id=user_id, project_id=project_id)
return bool(set(role_ids) & user_roles.reaching_ids_set)
def users_with_named_role(self, project_id, name):
""" returns in sorted order """
roles = self.project_roles(project_id)
return sorted(RoleCache(self, roles.find(name=name)).users_that_reach, key=lambda u:u.username)
def userids_with_named_role(self, project_id, name):
roles = self.project_roles(project_id)
return RoleCache(self, roles.find(name=name)).userids_that_reach
class RoleCache(object):
def __init__(self, cred, q):
self.cred = cred
self.q = q
def find(self, **kw):
tests = kw.items()
def _iter():
for r in self:
for k,v in tests:
val = getattr(r, k)
if callable(v):
if not v(val): break
elif v != val: break
else:
yield r
return RoleCache(self.cred, _iter())
def get(self, **kw):
for x in self.find(**kw): return x
return None
def __iter__(self):
return self.index.itervalues()
def __len__(self):
return len(self.index)
@LazyProperty
def index(self):
return dict((r._id, r) for r in self.q)
@LazyProperty
def named(self):
return RoleCache(self.cred, (
r for r in self
if r.name and not r.name.startswith('*')))
@LazyProperty
def reverse_index(self):
rev_index = defaultdict(list)
for r in self:
for rr_id in r.roles:
rev_index[rr_id].append(r)
return rev_index
@LazyProperty
def roles_that_reach(self):
def _iter():
visited = set()
to_visit = list(self)
while to_visit:
r = to_visit.pop(0)
if r in visited: continue
visited.add(r)
yield r
pr_rindex = self.cred.project_roles(r.project_id).reverse_index
to_visit += pr_rindex[r._id]
return RoleCache(self.cred, _iter())
@LazyProperty
def users_that_reach(self):
return [
r.user for r in self.roles_that_reach if r.user ]
@LazyProperty
def userids_that_reach(self):
return [
r.user_id for r in self.roles_that_reach ]
@LazyProperty
def reaching_roles(self):
def _iter():
to_visit = self.index.items()
visited = set()
while to_visit:
(rid, role) = to_visit.pop()
if rid in visited: continue
yield role
pr_index = self.cred.project_roles(role.project_id).index
if rid in pr_index:
for i in pr_index[rid].roles:
if i in pr_index:
to_visit.append((i, pr_index[i]))
return RoleCache(self.cred, _iter())
@LazyProperty
def reaching_ids(self):
return [ r._id for r in self.reaching_roles ]
@LazyProperty
def reaching_ids_set(self):
return set(self.reaching_ids)
def has_access(obj, permission, user=None, project=None):
'''Return whether the given user has the permission name on the given object.
- First, all the roles for a user in the given project context are computed.
- Next, for each role, the given object's ACL is examined linearly. If an ACE
is found which matches the permission and user, and that ACE ALLOWs access,
then the function returns True and access is permitted. If the ACE DENYs
access, then that role is removed from further consideration.
- If the obj is not a Neighborhood and the given user has the 'admin'
permission on the current neighborhood, then the function returns True and
access is allowed.
- If the obj is not a Project and the given user has the 'admin'
permission on the current project, then the function returns True and
access is allowed.
- If none of the ACEs on the object ALLOW access, and there are no more roles
to be considered, then the function returns False and access is denied.
- Processing continues using the remaining roles and the
obj.parent_security_context(). If the parent_security_context is None, then
the function returns False and access is denied.
The effect of this processing is that if *any* role for the user is ALLOWed
access via a linear traversal of the ACLs, then access is allowed. All of the
users roles must either be explicitly DENYed or processing terminate with no
matches to DENY access to the resource.
'''
from allura import model as M
def predicate(obj=obj, user=user, project=project, roles=None):
if obj is None:
return False
if roles is None:
if user is None: user = c.user
assert user, 'c.user should always be at least M.User.anonymous()'
cred = Credentials.get()
if project is None:
if isinstance(obj, M.Neighborhood):
project = obj.neighborhood_project
if project is None:
log.error('Neighborhood project missing for %s', obj)
return False
elif isinstance(obj, M.Project):
project = obj.root_project
else:
project = c.project.root_project
roles = cred.user_roles(user_id=user._id, project_id=project._id).reaching_ids
chainable_roles = []
for rid in roles:
for ace in obj.acl:
if M.ACE.match(ace, rid, permission):
if ace.access == M.ACE.ALLOW:
# access is allowed
# log.info('%s: True', txt)
return True
else:
# access is denied for this role
break
else:
# access neither allowed or denied, may chain to parent context
chainable_roles.append(rid)
parent = obj.parent_security_context()
if parent and chainable_roles:
result = has_access(parent, permission, user=user, project=project)(
roles=tuple(chainable_roles))
elif not isinstance(obj, M.Neighborhood):
result = has_access(project.neighborhood, 'admin', user=user)()
if not (result or isinstance(obj, M.Project)):
result = has_access(project, 'admin', user=user)()
else:
result = False
# log.info('%s: %s', txt, result)
return result
return TruthyCallable(predicate)
def require(predicate, message=None):
'''
Example: require(has_access(c.app, 'read'))
:param callable predicate: truth function to call
:param str message: message to show upon failure
:raises: HTTPForbidden or HTTPUnauthorized
'''
from allura import model as M
if predicate(): return
if not message:
message = """You don't have permission to do that.
You must ask a project administrator for rights to perform this task.
Please click the back button to return to the previous page."""
if c.user != M.User.anonymous():
request.environ['error_message'] = message
raise exc.HTTPForbidden(detail=message)
else:
raise exc.HTTPUnauthorized()
def require_access(obj, permission, **kwargs):
if obj is not None:
predicate = has_access(obj, permission, **kwargs)
return require(predicate, message='%s access required' % permission.capitalize())
else:
raise exc.HTTPForbidden(detail="Could not verify permissions for this page.")
def require_authenticated():
'''
:raises: HTTPUnauthorized if current user is anonymous
'''
from allura import model as M
if c.user == M.User.anonymous():
raise exc.HTTPUnauthorized()
def simple_grant(acl, role_id, permission):
from allura.model.types import ACE
for ace in acl:
if ace.role_id == role_id and ace.permission == permission: return
acl.append(ACE.allow(role_id, permission))
def simple_revoke(acl, role_id, permission):
remove = []
for i, ace in enumerate(acl):
if ace.role_id == role_id and ace.permission == permission:
remove.append(i)
for i in reversed(remove):
acl.pop(i)