| # -*- coding: utf-8 -*- |
| # |
| # Licensed to the Apache Software Foundation (ASF) under one |
| # or more contributor license agreements. See the NOTICE file |
| # distributed with this work for additional information |
| # regarding copyright ownership. The ASF licenses this file |
| # to you under the Apache License, Version 2.0 (the |
| # "License"); you may not use this file except in compliance |
| # with the License. You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, |
| # software distributed under the License is distributed on an |
| # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| # KIND, either express or implied. See the License for the |
| # specific language governing permissions and limitations |
| # under the License. |
| |
| """Tests for Apache(TM) Bloodhound's product permissions subsystem""" |
| |
| import unittest |
| |
| from trac.admin.api import AdminCommandError |
| from trac import perm |
| from trac.resource import Neighborhood |
| from trac.test import Mock |
| from trac.tests.perm import DefaultPermissionStoreTestCase,\ |
| PermissionSystemTestCase, PermissionCacheTestCase,\ |
| PermissionPolicyTestCase, TestPermissionPolicy, TestPermissionRequestor |
| |
| from multiproduct.api import MultiProductSystem |
| from multiproduct.env import ProductEnvironment |
| from multiproduct.model import Product |
| from multiproduct.perm import MultiproductPermissionPolicy, sudo |
| from tests.env import MultiproductTestCase |
| |
| # DefaultPermission policy has its own cache that causes |
| # test_product_trac_admin_actions to fail sometimes. |
| perm.DefaultPermissionPolicy.CACHE_EXPIRY = 0 |
| |
| |
| class ProductDefaultPermissionStoreTestCase(DefaultPermissionStoreTestCase, |
| MultiproductTestCase): |
| |
| def setUp(self): |
| self.global_env = self._setup_test_env() |
| self._upgrade_mp(self.global_env) |
| self._setup_test_log(self.global_env) |
| self._load_product_from_data(self.env, self.default_product) |
| self.env = ProductEnvironment(self.env, self.default_product) |
| |
| self.store = perm.DefaultPermissionStore(self.env) |
| |
| def test_env_isolation(self): |
| global_env = self.global_env |
| env = self.env |
| |
| self._load_product_from_data(self.global_env, 'tp2') |
| env1 = ProductEnvironment(self.global_env, 'tp2') |
| |
| global_store = perm.DefaultPermissionStore(global_env) |
| store = perm.DefaultPermissionStore(env) |
| store1 = perm.DefaultPermissionStore(env1) |
| |
| global_env.db_transaction.executemany( |
| "INSERT INTO permission VALUES (%s,%s)", |
| [('dev', 'WIKI_MODIFY'), |
| ('dev', 'REPORT_ADMIN'), |
| ('john', 'dev')]) |
| env.db_transaction.executemany( |
| "INSERT INTO permission VALUES (%s,%s)", |
| [('dev', 'WIKI_VIEW'), |
| ('dev', 'REPORT_VIEW'), |
| ('john', 'dev')]) |
| env1.db_transaction.executemany( |
| "INSERT INTO permission VALUES (%s,%s)", |
| [('dev', 'TICKET_CREATE'), |
| ('dev', 'MILESTONE_VIEW'), |
| ('john', 'dev')]) |
| |
| self.assertEquals(['REPORT_ADMIN', 'WIKI_MODIFY'], |
| sorted(global_store.get_user_permissions('john'))) |
| self.assertEquals(['REPORT_VIEW', 'WIKI_VIEW'], |
| sorted(store.get_user_permissions('john'))) |
| self.assertEquals(['MILESTONE_VIEW', 'TICKET_CREATE'], |
| sorted(store1.get_user_permissions('john'))) |
| |
| class ProductPermissionSystemTestCase(PermissionSystemTestCase, |
| MultiproductTestCase): |
| @property |
| def env(self): |
| env = getattr(self, '_env', None) |
| if env is None: |
| self.global_env = self._setup_test_env(enable=[ |
| perm.PermissionSystem, |
| perm.DefaultPermissionStore, |
| TestPermissionRequestor]) |
| self._upgrade_mp(self.global_env) |
| self._setup_test_log(self.global_env) |
| self._load_product_from_data(self.global_env, self.default_product) |
| self._env = env = ProductEnvironment( |
| self.global_env, self.default_product) |
| return env |
| |
| @env.setter |
| def env(self, value): |
| pass |
| |
| def test_all_permissions(self): |
| # PRODUCT_ADMIN meta-permission in product context |
| self.assertEqual({'EMAIL_VIEW': True, 'TRAC_ADMIN': True, |
| 'TEST_CREATE': True, 'TEST_DELETE': True, |
| 'TEST_MODIFY': True, 'TEST_ADMIN': True, |
| 'PRODUCT_ADMIN' : True}, |
| self.perm.get_user_permissions()) |
| |
| def test_expand_actions_iter_7467(self): |
| # Check that expand_actions works with iterators (#7467) |
| # PRODUCT_ADMIN meta-permission in product context |
| perms = set(['EMAIL_VIEW', 'TRAC_ADMIN', 'TEST_DELETE', 'TEST_MODIFY', |
| 'TEST_CREATE', 'TEST_ADMIN', 'PRODUCT_ADMIN']) |
| self.assertEqual(perms, self.perm.expand_actions(['TRAC_ADMIN'])) |
| self.assertEqual(perms, self.perm.expand_actions(iter(['TRAC_ADMIN']))) |
| |
| |
| class ProductPermissionCacheTestCase(PermissionCacheTestCase, |
| MultiproductTestCase): |
| @property |
| def env(self): |
| env = getattr(self, '_env', None) |
| if env is None: |
| self.global_env = self._setup_test_env(enable=[ |
| perm.DefaultPermissionStore, |
| perm.DefaultPermissionPolicy, |
| TestPermissionRequestor]) |
| self._upgrade_mp(self.global_env) |
| self._setup_test_log(self.global_env) |
| self._load_product_from_data(self.global_env, self.default_product) |
| self._env = env = ProductEnvironment( |
| self.global_env, self.default_product) |
| return env |
| |
| @env.setter |
| def env(self, value): |
| pass |
| |
| |
| class ProductNeighborhoodPermissionCacheTestCase(ProductPermissionCacheTestCase, |
| MultiproductTestCase): |
| @property |
| def env(self): |
| env = getattr(self, '_env', None) |
| if env is None: |
| self.global_env = self._setup_test_env(enable=[ |
| perm.DefaultPermissionStore, |
| perm.DefaultPermissionPolicy, |
| MultiProductSystem, |
| TestPermissionRequestor]) |
| self._upgrade_mp(self.global_env) |
| self._setup_test_log(self.global_env) |
| self._load_product_from_data(self.global_env, self.default_product) |
| self._env = env = ProductEnvironment( |
| self.global_env, self.default_product) |
| return env |
| |
| @env.setter |
| def env(self, value): |
| pass |
| |
| def setUp(self): |
| ProductPermissionCacheTestCase.setUp(self) |
| nbh = Neighborhood('product', self.default_product) |
| resource = nbh.child(None, None) |
| self.perm = perm.PermissionCache(self.global_env, 'testuser', resource) |
| |
| |
| class SudoTestCase(ProductPermissionCacheTestCase): |
| loader = unittest.defaultTestLoader |
| tcnames = loader.getTestCaseNames(ProductPermissionCacheTestCase) |
| _gen_tests = {} |
| |
| def test_sudo_wrong_context(self): |
| sudoperm = sudo(None, 'EMAIL_VIEW', ['TEST_ADMIN']) |
| |
| with self.assertRaises(RuntimeError) as test_cm: |
| sudoperm.has_permission('TEST_MODIFY') |
| self.assertEqual('Permission check out of context', |
| str(test_cm.exception)) |
| |
| with self.assertRaises(ValueError) as test_cm: |
| with sudoperm: |
| pass |
| self.assertEquals('Context manager not bound to request object', |
| str(test_cm.exception)) |
| |
| def test_sudo_fail_require(self): |
| sudoperm = sudo(None, 'EMAIL_VIEW', ['TEST_ADMIN']) |
| |
| sudoperm.perm = self.perm |
| with self.assertRaises(perm.PermissionError) as test_cm: |
| sudoperm.require('TRAC_ADMIN') |
| self.assertEqual('EMAIL_VIEW', test_cm.exception.action) |
| |
| def test_sudo_grant_meta_perm(self): |
| self.env.parent.enable_component(perm.PermissionSystem) |
| self.env.enable_component(perm.PermissionSystem) |
| del self.env.parent.enabled[perm.PermissionSystem] |
| del self.env.enabled[perm.PermissionSystem] |
| |
| sudoperm = sudo(None, 'TEST_CREATE', ['TRAC_ADMIN']) |
| sudoperm.perm = self.perm |
| |
| self.assertTrue(sudoperm.has_permission('EMAIL_VIEW')) |
| |
| def test_sudo_ambiguous(self): |
| with self.assertRaises(ValueError) as test_cm: |
| sudo(None, 'TEST_MODIFY', ['TEST_MODIFY', 'TEST_DELETE'], |
| ['TEST_MODIFY', 'TEST_CREATE']) |
| self.assertEquals('Impossible to grant and revoke (TEST_MODIFY)', |
| str(test_cm.exception)) |
| |
| with self.assertRaises(ValueError) as test_cm: |
| sudoperm = sudo(None, 'TEST_MODIFY', ['TEST_ADMIN'], |
| ['TEST_MODIFY', 'TEST_CREATE']) |
| sudoperm.perm = self.perm |
| self.assertEquals('Impossible to grant and revoke ' |
| '(TEST_CREATE, TEST_MODIFY)', |
| str(test_cm.exception)) |
| |
| with self.assertRaises(ValueError) as test_cm: |
| req = Mock(perm=self.perm) |
| sudo(req, 'TEST_MODIFY', ['TEST_ADMIN'], |
| ['TEST_MODIFY', 'TEST_CREATE']) |
| self.assertEquals('Impossible to grant and revoke ' |
| '(TEST_CREATE, TEST_MODIFY)', |
| str(test_cm.exception)) |
| |
| # Sudo permission context equivalent to permissions cache |
| # if there's no action to require, allow or deny. |
| def _test_with_sudo_rules(tcnm, prefix, grant): |
| target = getattr(ProductPermissionCacheTestCase, tcnm) |
| |
| def _sudo_eq_checker(self): |
| for action in grant: |
| self.perm_system.revoke_permission('testuser', action) |
| realperm = self.perm |
| self.perm = sudo(None, [], grant, []) |
| self.perm.perm = realperm |
| target(self) |
| |
| _sudo_eq_checker.func_name = prefix + tcnm |
| return _sudo_eq_checker |
| |
| for tcnm in tcnames: |
| f1 = _test_with_sudo_rules(tcnm, '', []) |
| f2 = _test_with_sudo_rules(tcnm, 'test_sudo_partial_', |
| ['TEST_MODIFY']) |
| f3 = _test_with_sudo_rules(tcnm, 'test_sudo_full_', |
| ['TEST_MODIFY', 'TEST_ADMIN']) |
| for f in (f1, f2, f3): |
| _gen_tests[f.func_name] = f |
| |
| del loader, tcnames, tcnm, f1, f2, f3 |
| |
| list(setattr(SudoTestCase, tcnm, f) |
| for tcnm, f in SudoTestCase._gen_tests.iteritems()) |
| |
| |
| class ProductPermissionPolicyTestCase(PermissionPolicyTestCase, |
| MultiproductTestCase): |
| @property |
| def env(self): |
| env = getattr(self, '_env', None) |
| if env is None: |
| self.global_env = self._setup_test_env(enable=[ |
| perm.DefaultPermissionStore, |
| perm.DefaultPermissionPolicy, |
| perm.PermissionSystem, |
| TestPermissionPolicy, |
| TestPermissionRequestor, |
| MultiproductPermissionPolicy]) |
| self._upgrade_mp(self.global_env) |
| self._setup_test_log(self.global_env) |
| self._load_product_from_data(self.global_env, self.default_product) |
| self._env = env = ProductEnvironment( |
| self.global_env, self.default_product) |
| return env |
| |
| @env.setter |
| def env(self, value): |
| pass |
| |
| def setUp(self): |
| super(ProductPermissionPolicyTestCase, self).setUp() |
| |
| self.global_env.config.set('trac', 'permission_policies', |
| 'DefaultPermissionPolicy') |
| self.permsys = perm.PermissionSystem(self.env) |
| self.global_perm_admin = perm.PermissionAdmin(self.global_env) |
| self.product_perm_admin = perm.PermissionAdmin(self.env) |
| |
| def tearDown(self): |
| self.global_env.reset_db() |
| self.global_env = self.env = None |
| |
| def test_prepend_mp_policy(self): |
| self.assertEqual([MultiproductPermissionPolicy(self.env), self.policy], |
| self.permsys.policies) |
| |
| def test_policy_chaining(self): |
| self.env.config.set('trac', 'permission_policies', |
| 'TestPermissionPolicy,DefaultPermissionPolicy') |
| self.policy.grant('testuser', ['TEST_MODIFY']) |
| system = perm.PermissionSystem(self.env) |
| system.grant_permission('testuser', 'TEST_ADMIN') |
| |
| self.assertEqual(list(system.policies), |
| [MultiproductPermissionPolicy(self.env), |
| self.policy, |
| perm.DefaultPermissionPolicy(self.env)]) |
| self.assertEqual('TEST_MODIFY' in self.perm, True) |
| self.assertEqual('TEST_ADMIN' in self.perm, True) |
| self.assertEqual(self.policy.results, |
| {('testuser', 'TEST_MODIFY'): True, |
| ('testuser', 'TEST_ADMIN'): None}) |
| |
| def test_product_trac_admin_success(self): |
| """TRAC_ADMIN in global env also valid in product env |
| """ |
| self.global_perm_admin._do_add('testuser', 'TRAC_ADMIN') |
| self.assertTrue(self.perm.has_permission('TRAC_ADMIN')) |
| |
| def test_product_trac_admin_actions(self): |
| """Allow all actions in product scope for TRAC_ADMIN |
| """ |
| self.global_perm_admin._do_add('testuser', 'TRAC_ADMIN') |
| |
| all_actions = self.permsys.get_actions() |
| self.assertEquals(['TEST_CREATE', 'EMAIL_VIEW', 'TRAC_ADMIN', |
| 'TEST_DELETE', 'TEST_MODIFY', 'PRODUCT_ADMIN', |
| 'TEST_ADMIN'], all_actions) |
| self.assertEquals({}, self.permsys.get_user_permissions('testuser')) |
| for action in all_actions: |
| self.assertTrue(self.perm.has_permission(action), |
| 'Check for permission action %s' % (action,)) |
| self.assertFalse(self.perm.has_permission('UNKNOWN_PERM')) |
| |
| # Clear permissions cache and retry |
| self.perm._cache.clear() |
| self.global_perm_admin._do_remove('testuser', 'TRAC_ADMIN') |
| |
| all_actions = self.permsys.get_actions() |
| self.assertEquals(['TEST_CREATE', 'EMAIL_VIEW', 'TRAC_ADMIN', |
| 'TEST_DELETE', 'TEST_MODIFY', 'PRODUCT_ADMIN', |
| 'TEST_ADMIN'], all_actions) |
| self.assertEquals({}, self.permsys.get_user_permissions('testuser')) |
| for action in all_actions: |
| self.assertFalse(self.perm.has_permission(action), |
| 'Check for permission action %s' % (action,)) |
| |
| def test_product_trac_admin_fail_local(self): |
| """TRAC_ADMIN granted in product env will be ignored |
| """ |
| try: |
| # Not needed but added just in case , also for readability |
| self.global_perm_admin._do_remove('testuser', 'TRAC_ADMIN') |
| except AdminCommandError: |
| pass |
| |
| # Setting TRAC_ADMIN permission in product scope is in vain |
| # since it controls access to critical actions affecting the whole site |
| # This will protect the system against malicious actors |
| # and / or failures leading to the addition of TRAC_ADMIN permission |
| # in product perm store in spite of obtaining unrighteous super powers. |
| # On the other hand this also means that PRODUCT_ADMIN(s) are |
| # able to set user permissions at will without jeopardizing system |
| # integrity and stability. |
| self.product_perm_admin._do_add('testuser', 'TRAC_ADMIN') |
| self.assertFalse(self.perm.has_permission('TRAC_ADMIN')) |
| |
| def test_product_owner_perm(self): |
| """Product owner automatically granted with PRODUCT_ADMIN |
| """ |
| self.assertIs(self.env.product.owner, None) |
| self.assertFalse(self.perm.has_permission('PRODUCT_ADMIN')) |
| |
| self.env.product.owner = 'testuser' |
| # FIXME: update really needed ? |
| self.env.product.update() |
| try: |
| # Not needed but added just in case , also for readability |
| self.global_perm_admin._do_remove('testuser', 'TRAC_ADMIN') |
| except AdminCommandError: |
| pass |
| self.perm._cache.clear() |
| |
| self.assertTrue(self.perm.has_permission('PRODUCT_ADMIN')) |
| self.assertFalse(self.perm.has_permission('TRAC_ADMIN')) |
| |
| def test_new_product_perm(self): |
| """Only product owner and TRAC_ADMIN will access new product |
| """ |
| newproduct = Product(self.global_env) |
| newproduct.prefix = 'NEW' |
| newproduct.name = 'New product' |
| newproduct.owner = 'owneruser' |
| newproduct.insert() |
| |
| env = ProductEnvironment(self.global_env, newproduct) |
| self.global_perm_admin._do_add('adminuser', 'TRAC_ADMIN') |
| admin_perm = perm.PermissionCache(env, 'adminuser') |
| owner_perm = perm.PermissionCache(env, 'owneruser') |
| user_perm = perm.PermissionCache(env, 'testuser') |
| global_permsys = perm.PermissionSystem(self.global_env) |
| permsys = perm.PermissionSystem(env) |
| |
| self.assertEquals({'EMAIL_VIEW': True, 'TEST_ADMIN': True, |
| 'TEST_CREATE': True, 'TEST_DELETE': True, |
| 'TEST_MODIFY': True, 'TRAC_ADMIN' : True}, |
| global_permsys.get_user_permissions('adminuser')) |
| self.assertEquals({}, global_permsys.get_user_permissions('owneruser')) |
| self.assertEquals({}, global_permsys.get_user_permissions('testuser')) |
| self.assertEquals({}, permsys.get_user_permissions('adminuser')) |
| self.assertEquals({}, permsys.get_user_permissions('owneruser')) |
| self.assertEquals({}, permsys.get_user_permissions('testuser')) |
| |
| all_actions = self.permsys.get_actions() |
| all_actions.remove('TRAC_ADMIN') |
| for action in all_actions: |
| self.assertTrue(admin_perm.has_permission(action)) |
| self.assertTrue(owner_perm.has_permission(action)) |
| self.assertFalse(user_perm.has_permission(action)) |
| |
| self.assertTrue(admin_perm.has_permission('TRAC_ADMIN')) |
| self.assertFalse(owner_perm.has_permission('TRAC_ADMIN')) |
| self.assertFalse(user_perm.has_permission('TRAC_ADMIN')) |
| |
| |
| def test_suite(): |
| suite = unittest.TestSuite() |
| suite.addTest(unittest.makeSuite(ProductDefaultPermissionStoreTestCase, |
| 'test')) |
| suite.addTest(unittest.makeSuite(ProductPermissionSystemTestCase, 'test')) |
| suite.addTest(unittest.makeSuite(ProductPermissionCacheTestCase, 'test')) |
| suite.addTest(unittest.makeSuite(ProductNeighborhoodPermissionCacheTestCase, |
| 'test')) |
| suite.addTest(unittest.makeSuite(ProductPermissionPolicyTestCase, 'test')) |
| |
| suite.addTest(unittest.makeSuite(SudoTestCase, 'test')) |
| return suite |
| |
| if __name__ == '__main__': |
| unittest.main(defaultTest='test_suite') |
| |
