zlwu
0bddf38e52
Separate cross-team selection from attribution maintenance and apply the same scope to options, history filters and writes. Add an opt-in tenant permission without granting existing roles. Protect authorization mutation paths against self-grants, validate tenant targets and links, and include menu ancestors atomically. Cover role union, revocation, mobile writes and administration boundaries with regression tests.
177 lines
9.5 KiB
Python
177 lines
9.5 KiB
Python
"""Project choices and writes share explicit team/all-project permission boundaries."""
|
|
import unittest
|
|
|
|
from fastapi import HTTPException
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.base import Base
|
|
from app.models.models import (BizCollectionProtocol, SysMenu, SysRole, SysRoleMenu,
|
|
SysUserRole, SysPost, SysPostRole, SysUserPost)
|
|
from app.models.project_management import Project, ProjectTeam, ProjectTeamMember
|
|
from app.services import project_management as svc
|
|
from app.services.project_assignment import create_assignment
|
|
|
|
|
|
class ProjectSelectionPermissionsTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.engine = create_engine('sqlite:///:memory:')
|
|
Base.metadata.create_all(self.engine)
|
|
self.db = Session(self.engine)
|
|
self.user = dict(user_id='member', role='admin', tenant_no='t')
|
|
kinds = ','.join(svc.COUNT_TYPES)
|
|
self.db.add_all([
|
|
ProjectTeam(id='own', tenant_no='t', team_name='Own', leader_user_id='leader'),
|
|
ProjectTeam(id='other', tenant_no='t', team_name='Other', leader_user_id='other'),
|
|
ProjectTeam(id='foreign', tenant_no='foreign', team_name='Foreign', leader_user_id='member'),
|
|
ProjectTeamMember(id='member', tenant_no='t', team_id='own', user_id='member'),
|
|
Project(id='own-p', tenant_no='t', project_no='OWN', project_name='Own', team_id='own', leader_user_id='other', status='active', biz_types=kinds),
|
|
Project(id='other-p', tenant_no='t', project_no='OTHER', project_name='Other', team_id='other', leader_user_id='other', status='active', biz_types=kinds),
|
|
Project(id='foreign-p', tenant_no='foreign', project_no='FOREIGN', project_name='Foreign', team_id='foreign', leader_user_id='member', status='active', biz_types=kinds),
|
|
])
|
|
self.db.commit()
|
|
|
|
def tearDown(self):
|
|
self.db.close()
|
|
self.engine.dispose()
|
|
|
|
def grant(self, code, suffix='all', *, inherited=False):
|
|
self.db.add_all([
|
|
SysRole(id=suffix, tenant_no='t', role_name=suffix, data_scope='4'),
|
|
SysMenu(id=suffix, tenant_no='t', menu_name=suffix, menu_type='F', visible='0', perms=code),
|
|
SysRoleMenu(id=suffix, tenant_no='t', role_id=suffix, menu_id=suffix),
|
|
])
|
|
if inherited:
|
|
self.db.add_all([
|
|
SysPost(id=suffix, tenant_no='t', post_name=suffix, post_code=suffix),
|
|
SysPostRole(id=suffix, tenant_no='t', post_id=suffix, role_id=suffix),
|
|
SysUserPost(id=suffix, tenant_no='t', user_id='member', post_id=suffix),
|
|
])
|
|
else:
|
|
self.db.add(SysUserRole(id=suffix, tenant_no='t', user_id='member', role_id=suffix))
|
|
self.db.commit()
|
|
|
|
def choices(self, user=None, kind='cell_storage'):
|
|
return {r['id'] for r in svc.options(self.db, 't', user or self.user, kind)['list']}
|
|
|
|
def assert_forbidden(self, project_id='other-p', kind='cell_storage'):
|
|
with self.assertRaises(HTTPException) as caught:
|
|
create_assignment(self.db, 't', self.user, project_id, kind)
|
|
self.assertEqual(caught.exception.status_code, 403)
|
|
|
|
def test_project_leader_alone_does_not_bypass_team_membership(self):
|
|
self.db.get(Project, 'other-p').leader_user_id = 'member'
|
|
self.db.commit()
|
|
self.assertEqual(self.choices(), {'own-p'})
|
|
self.assert_forbidden()
|
|
|
|
def test_team_leader_and_membership_union(self):
|
|
self.assertEqual(self.choices(dict(user_id='leader', role='admin')), {'own-p'})
|
|
self.db.add(ProjectTeamMember(id='second', tenant_no='t', team_id='other', user_id='member'))
|
|
self.db.commit()
|
|
self.assertEqual(self.choices(), {'own-p', 'other-p'})
|
|
self.db.get(ProjectTeamMember, 'second').del_flag = 1
|
|
self.db.commit()
|
|
self.assertEqual(self.choices(), {'own-p'})
|
|
self.assert_forbidden()
|
|
|
|
def test_admin_with_no_team_has_no_choices(self):
|
|
self.assertEqual(self.choices(dict(user_id='outsider', role='admin')), set())
|
|
|
|
def test_assignment_edit_does_not_grant_all_projects(self):
|
|
self.grant('project:assignment:edit', 'assignment')
|
|
self.assertEqual(self.choices(), {'own-p'})
|
|
self.assert_forbidden()
|
|
|
|
def test_all_permission_in_role_union_allows_all_business_types(self):
|
|
self.grant('project:assignment:edit', 'assignment')
|
|
self.grant('project:project:all')
|
|
for kind in svc.BUSINESS:
|
|
with self.subTest(kind=kind):
|
|
self.assertEqual(self.choices(kind=kind), {'own-p', 'other-p'})
|
|
self.assertEqual(create_assignment(self.db, 't', self.user, 'other-p', kind), 'other-p')
|
|
with self.assertRaises(HTTPException) as caught:
|
|
create_assignment(self.db, 't', self.user, 'foreign-p', 'cell_storage')
|
|
self.assertEqual(caught.exception.status_code, 404)
|
|
|
|
def test_all_permission_does_not_bypass_project_state_or_business_type(self):
|
|
self.grant('project:project:all')
|
|
self.db.get(Project, 'other-p').status = 'disabled'
|
|
self.db.commit()
|
|
self.assertEqual(self.choices(), {'own-p'})
|
|
with self.assertRaises(HTTPException):
|
|
create_assignment(self.db, 't', self.user, 'other-p', 'cell_storage')
|
|
self.db.get(Project, 'other-p').status = 'active'
|
|
self.db.get(Project, 'other-p').biz_types = 'basic_experiment'
|
|
self.db.commit()
|
|
self.assertEqual(self.choices(), {'own-p'})
|
|
with self.assertRaises(HTTPException):
|
|
create_assignment(self.db, 't', self.user, 'other-p', 'cell_storage')
|
|
|
|
def test_permission_revocation_is_effective_on_next_request(self):
|
|
self.grant('project:project:all')
|
|
self.assertEqual(self.choices(), {'own-p', 'other-p'})
|
|
for model, field, disabled, enabled in [
|
|
(SysRole, 'status', '0', '1'), (SysRoleMenu, 'del_flag', 1, 0),
|
|
(SysMenu, 'status', '0', '1'), (SysUserRole, 'del_flag', 1, 0),
|
|
]:
|
|
with self.subTest(model=model.__name__):
|
|
setattr(self.db.get(model, 'all'), field, disabled)
|
|
self.db.commit()
|
|
self.assertEqual(self.choices(), {'own-p'})
|
|
self.assert_forbidden()
|
|
setattr(self.db.get(model, 'all'), field, enabled)
|
|
self.db.commit()
|
|
|
|
def test_post_inherited_all_permission_and_disabled_post(self):
|
|
self.grant('project:project:all', inherited=True)
|
|
self.assertEqual(self.choices(), {'own-p', 'other-p'})
|
|
self.assertEqual(create_assignment(self.db, 't', self.user, 'other-p', 'presign'), 'other-p')
|
|
self.db.get(SysPost, 'all').status = '0'
|
|
self.db.commit()
|
|
self.assertEqual(self.choices(), {'own-p'})
|
|
self.assert_forbidden()
|
|
|
|
def test_filter_options_require_team_scope_and_preserve_business_scope(self):
|
|
self.grant('project:assignment:edit', 'assignment')
|
|
self.db.add_all([
|
|
BizCollectionProtocol(id='own-record', protocol_no='OWN', tenant_no='t', create_by='member', project_id='own-p'),
|
|
BizCollectionProtocol(id='other-record', protocol_no='OTHER', tenant_no='t', create_by='member', project_id='other-p'),
|
|
BizCollectionProtocol(id='hidden-record', protocol_no='HIDDEN', tenant_no='t', create_by='other', project_id='own-p'),
|
|
])
|
|
self.db.commit()
|
|
def filters(**kwargs):
|
|
return {r['id'] for r in svc.filter_options(self.db, 't', self.user, 'cell_storage', **kwargs)['list']}
|
|
self.assertEqual(filters(), {'own-p'})
|
|
self.grant('project:project:all')
|
|
self.assertEqual(filters(), {'own-p', 'other-p'})
|
|
self.assertEqual(svc.records(self.db, 't', self.user, 'cell_storage')['total'], 2)
|
|
# Mobile passes its stricter business query to the same options service.
|
|
mobile_scope = self.db.query(BizCollectionProtocol).filter(BizCollectionProtocol.id == 'own-record')
|
|
self.assertEqual(filters(business_scope=mobile_scope), {'own-p'})
|
|
|
|
def test_cross_group_reassignment_needs_both_permissions(self):
|
|
self.grant('project:assignment:edit', 'assignment')
|
|
self.db.add(BizCollectionProtocol(id='record', protocol_no='R', tenant_no='t', create_by='member', project_id='own-p'))
|
|
self.db.commit()
|
|
with self.assertRaises(HTTPException) as caught:
|
|
svc.preview_assignment(self.db, 't', self.user, 'cell_storage', 'record', 'other-p')
|
|
self.assertEqual(caught.exception.status_code, 403)
|
|
self.grant('project:project:all')
|
|
preview = svc.preview_assignment(self.db, 't', self.user, 'cell_storage', 'record', 'other-p')
|
|
self.db.get(SysRoleMenu, 'all').del_flag = 1
|
|
self.db.commit()
|
|
with self.assertRaises(HTTPException) as caught:
|
|
svc.assign(self.db, 't', self.user, 'cell_storage', 'record', 'other-p', 'reason', preview['records'])
|
|
self.assertEqual(caught.exception.status_code, 403)
|
|
self.assertEqual(self.db.get(BizCollectionProtocol, 'record').project_id, 'own-p')
|
|
|
|
def test_all_permission_does_not_grant_maintenance_or_management_actions(self):
|
|
self.grant('project:project:all')
|
|
for code in ('project:assignment:edit', 'project:project:edit', 'project:team:edit'):
|
|
with self.subTest(code=code), self.assertRaises(HTTPException):
|
|
svc.check_permission(self.db, 't', self.user, code)
|
|
|
|
def test_super_admin_retains_tenant_boundary(self):
|
|
self.assertEqual(self.choices(dict(user_id='root', role='super_admin')), {'own-p', 'other-p'})
|