This is an automated email from the ASF dual-hosted git repository. asf-gitbox-commits pushed a commit to branch db/8607 in repository https://gitbox.apache.org/repos/asf/allura.git
commit 2363e20f9ebdf74c5c744bf57aae0f96fb9b078c Author: Dave Brondsema <[email protected]> AuthorDate: Fri May 15 18:03:43 2026 -0400 [#8607] check project when loading role_id inputs --- Allura/allura/app.py | 4 +-- Allura/allura/ext/admin/admin_main.py | 43 +++++++++++++++++++--------- Allura/allura/tests/functional/test_admin.py | 16 +++++++++++ 3 files changed, 47 insertions(+), 16 deletions(-) diff --git a/Allura/allura/app.py b/Allura/allura/app.py index ac602fe50..78798f7f5 100644 --- a/Allura/allura/app.py +++ b/Allura/allura/app.py @@ -1062,7 +1062,7 @@ def update(self, card=None, **kw): del_group_ids.append(str(acl['role_id'])) def get_role(_id): - return model.ProjectRole.query.get(_id=ObjectId(_id)) + return model.ProjectRole.query.get(_id=ObjectId(_id), project_id=c.project.root_project._id) groups = list(map(get_role, group_ids)) new_groups = list(map(get_role, new_group_ids)) del_groups = list(map(get_role, del_group_ids)) @@ -1077,7 +1077,7 @@ def group_names(groups): group_names(groups + new_groups), self.app.config.options['mount_point'])) - role_ids = list(map(ObjectId, group_ids + new_group_ids)) + role_ids = [g._id for g in groups + new_groups if g is not None] self.app.config.acl += [ model.ACE.allow(r, perm) for r in role_ids] diff --git a/Allura/allura/ext/admin/admin_main.py b/Allura/allura/ext/admin/admin_main.py index 62b8132c7..cf97d090e 100644 --- a/Allura/allura/ext/admin/admin_main.py +++ b/Allura/allura/ext/admin/admin_main.py @@ -31,6 +31,7 @@ from tg.decorators import with_trailing_slash, without_trailing_slash from webob import exc from bson import ObjectId +from bson.errors import InvalidId from ming.odm.odmsession import ThreadLocalODMSession from ming.odm import session @@ -1059,14 +1060,18 @@ def update(self, card=None, **kw): role_ids = list(map(ObjectId, group_ids + new_group_ids)) permissions[perm] = role_ids c.project.acl = [] + + def role_names(roles): + return ','.join(sorted(pr.name for pr in roles)) + for perm, role_ids in permissions.items(): - def role_names(ids): return ','.join(sorted( - pr.name for pr in M.ProjectRole.query.find(dict(_id={'$in': ids})))) old_role_ids = old_permissions.get(perm, []) + roles = M.ProjectRole.query.find(dict(_id={'$in': role_ids}, project_id=c.project.root_project._id)).all() if old_role_ids != role_ids: + old_roles = M.ProjectRole.query.find(dict(_id={'$in': old_role_ids})).all() # no project_id check needed for old M.AuditLog.log('updated "%s" permissions: "%s" => "%s"', - perm, role_names(old_role_ids), role_names(role_ids)) - c.project.acl += [M.ACE.allow(rid, perm) for rid in role_ids] + perm, role_names(old_roles), role_names(roles)) + c.project.acl += [M.ACE.allow(role._id, perm) for role in roles] g.post_event('project_updated') redirect('.') @@ -1160,22 +1165,32 @@ def index(self, **kw): return dict(roles=roles, permissions_by_role=permissions_by_role, auth_role=auth_role, anon_role=anon_role) + def _role_for_current_project(self, role_id): + try: + role_oid = ObjectId(role_id) + except (InvalidId, TypeError): + return None + return M.ProjectRole.query.get( + _id=role_oid, + project_id=c.project.root_project._id) + @without_trailing_slash @expose('json:') @require_post() @h.vardec def change_perm(self, role_id, permission, allow="true", **kw): + role = self._role_for_current_project(role_id) + if not role: + return dict(error='Could not find group with id %s' % role_id) if allow == "true": - M.AuditLog.log('granted permission %s to group %s', permission, - M.ProjectRole.query.get(_id=ObjectId(role_id)).name) - c.project.acl.append(M.ACE.allow(ObjectId(role_id), permission)) + M.AuditLog.log('granted permission %s to group %s', permission, role.name) + c.project.acl.append(M.ACE.allow(role._id, permission)) else: admin_group_id = str(M.ProjectRole.by_name('Admin')._id) if admin_group_id == role_id and permission == 'admin': return dict(error='You cannot remove the admin permission from the admin group.') - M.AuditLog.log('revoked permission %s from group %s', permission, - M.ProjectRole.query.get(_id=ObjectId(role_id)).name) - c.project.acl.remove(M.ACE.allow(ObjectId(role_id), permission)) + M.AuditLog.log('revoked permission %s from group %s', permission, role.name) + c.project.acl.remove(M.ACE.allow(role._id, permission)) g.post_event('project_updated') return self._map_group_permissions() @@ -1186,7 +1201,7 @@ def change_perm(self, role_id, permission, allow="true", **kw): def add_user(self, role_id, username, **kw): if not username or username == '*anonymous': return dict(error='You must choose a user to add.') - group = M.ProjectRole.query.get(_id=ObjectId(role_id)) + group = self._role_for_current_project(role_id) user = M.User.query.get(username=username.strip(), pending=False) if not group: @@ -1209,12 +1224,12 @@ def add_user(self, role_id, username, **kw): @require_post() @h.vardec def remove_user(self, role_id, username, **kw): - group = M.ProjectRole.query.get(_id=ObjectId(role_id)) + group = self._role_for_current_project(role_id) + if not group: + return dict(error='Could not find group with id %s' % role_id) user = M.User.by_username(username.strip()) if group.name == 'Admin' and len(group.users_with_role()) == 1: return dict(error='You must have at least one user with the Admin role.') - if not group: - return dict(error='Could not find group with id %s' % role_id) if not user: return dict(error='User %s not found' % username) user_role = M.ProjectRole.by_user(user) diff --git a/Allura/allura/tests/functional/test_admin.py b/Allura/allura/tests/functional/test_admin.py index 49c590ba5..a87eb6d42 100644 --- a/Allura/allura/tests/functional/test_admin.py +++ b/Allura/allura/tests/functional/test_admin.py @@ -923,6 +923,22 @@ def test_permission_inherit(self): assert {'text': 'Does not have permission create', 'has': 'no', 'name': 'create'} in r.json[anon_id] + def test_add_user_rejects_foreign_role(self): + # role_id from a different project must not be usable to grant membership in this project + h.set_context('test2', neighborhood='Projects') + foreign_role_id = M.ProjectRole.by_name('Admin')._id + h.set_context('test', neighborhood='Projects') + + self.app.post('/admin/groups/add_user', params={ + 'role_id': str(foreign_role_id), + 'username': 'test-user-1'}) + + h.set_context('test', neighborhood='Projects') + user = M.User.by_username('test-user-1') + user_role = M.ProjectRole.by_user(user) + assert user_role is None or foreign_role_id not in user_role.roles, \ + 'foreign role_id was added to user.roles' + def test_admin_extension_sidebar(self): class FooSettingsController:
