Tests for rbac code

4 files changed, 191 insertions, 0 deletions

diff --git a/nova/auth/users.py b/nova/auth/users.py
index 98462b362..f3248166c 100644
--- a/nova/auth/users.py
+++ b/nova/auth/users.py
@@ -58,6 +58,10 @@ flags.DEFINE_string('role_ldap_subtree', 'ou=Groups,dc=example,dc=com', 'OU for
 # mapping with these flags is necessary because we're going to tie in to an existing ldap schema
 flags.DEFINE_string('ldap_cloudadmin',
     'cn=cloudadmins,ou=Groups,dc=example,dc=com', 'cn for Cloud Admins')
+flags.DEFINE_string('ldap_sysadmin',
+    'cn=sysadmins,ou=Groups,dc=example,dc=com', 'cn for Sysadmins')
+flags.DEFINE_string('ldap_netadmin',
+    'cn=netadmins,ou=Groups,dc=example,dc=com', 'cn for NetAdmins')
 
 # a user with one of these roles will be a superuser and have access to all api commands
 flags.DEFINE_list('superuser_roles', ['cloudadmin'], 'roles that ignore rbac checking completely')
diff --git a/nova/tests/access_unittest.py b/nova/tests/access_unittest.py
new file mode 100644
index 000000000..a1b574dd8
--- /dev/null
+++ b/nova/tests/access_unittest.py
@@ -0,0 +1,160 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+# Copyright [2010] [Anso Labs, LLC]
+#
+#    Licensed 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.
+
+import unittest
+
+
+from nova.auth.users import UserManager
+from nova.auth import rbac
+from nova import exception
+from nova import flags
+from nova import test
+
+FLAGS = flags.FLAGS
+class Context(object):
+    pass
+
+class AccessTestCase(test.BaseTestCase):
+    def setUp(self):
+        FLAGS.fake_libvirt = True
+        FLAGS.fake_storage = True
+        um = UserManager.instance()
+        # Make test users
+        try:
+            self.testadmin = um.create_user('testadmin')
+        except: pass
+        try:
+            self.testpmsys = um.create_user('testpmsys')
+        except: pass
+        try:
+            self.testnet = um.create_user('testnet')
+        except: pass
+        try:
+            self.testsys = um.create_user('testsys')
+        except: pass
+        # Assign some rules
+        try:
+            um.add_role('testadmin', 'cloudadmin')
+        except: pass
+        try:
+            um.add_role('testpmsys', 'sysadmin')
+        except: pass
+        try:
+            um.add_role('testnet', 'netadmin')
+        except: pass
+        try:
+            um.add_role('testsys', 'sysadmin')
+        except: pass
+
+        # Make a test project
+        try:
+            self.project = um.create_project('testproj', 'testpmsys', 'a test project', ['testpmsys', 'testnet', 'testsys'])
+        except: pass
+        try:
+            self.project.add_role(self.testnet, 'netadmin')
+        except: pass
+        try:
+            self.project.add_role(self.testsys, 'sysadmin')
+        except: pass
+        self.context = Context()
+        self.context.project = self.project
+        #user is set in each test
+        super(AccessTestCase, self).setUp()
+
+    def tearDown(self):
+        um = UserManager.instance()
+        # Delete the test project
+        um.delete_project('testproj')
+        # Delete the test user
+        um.delete_user('testadmin')
+        um.delete_user('testpmsys')
+        um.delete_user('testnet')
+        um.delete_user('testsys')
+        super(AccessTestCase, self).tearDown()
+
+    def test_001_allow_all(self):
+        self.context.user = self.testadmin
+        self.assertTrue(self._allow_all(self.context))
+        self.context.user = self.testpmsys
+        self.assertTrue(self._allow_all(self.context))
+        self.context.user = self.testnet
+        self.assertTrue(self._allow_all(self.context))
+        self.context.user = self.testsys
+        self.assertTrue(self._allow_all(self.context))
+
+    def test_002_allow_none(self):
+        self.context.user = self.testadmin
+        self.assertTrue(self._allow_none(self.context))
+        self.context.user = self.testpmsys
+        self.assertRaises(exception.NotAuthorized, self._allow_none, self.context)
+        self.context.user = self.testnet
+        self.assertRaises(exception.NotAuthorized, self._allow_none, self.context)
+        self.context.user = self.testsys
+        self.assertRaises(exception.NotAuthorized, self._allow_none, self.context)
+
+    def test_003_allow_project_manager(self):
+        self.context.user = self.testadmin
+        self.assertTrue(self._allow_project_manager(self.context))
+        self.context.user = self.testpmsys
+        self.assertTrue(self._allow_project_manager(self.context))
+        self.context.user = self.testnet
+        self.assertRaises(exception.NotAuthorized, self._allow_project_manager, self.context)
+        self.context.user = self.testsys
+        self.assertRaises(exception.NotAuthorized, self._allow_project_manager, self.context)
+
+    def test_004_allow_sys_and_net(self):
+        self.context.user = self.testadmin
+        self.assertTrue(self._allow_sys_and_net(self.context))
+        self.context.user = self.testpmsys # doesn't have the per project sysadmin
+        self.assertRaises(exception.NotAuthorized, self._allow_sys_and_net, self.context)
+        self.context.user = self.testnet
+        self.assertTrue(self._allow_sys_and_net(self.context))
+        self.context.user = self.testsys
+        self.assertTrue(self._allow_sys_and_net(self.context))
+
+    def test_005_allow_sys_no_pm(self):
+        self.context.user = self.testadmin
+        self.assertTrue(self._allow_sys_no_pm(self.context))
+        self.context.user = self.testpmsys
+        self.assertRaises(exception.NotAuthorized, self._allow_sys_no_pm, self.context)
+        self.context.user = self.testnet
+        self.assertRaises(exception.NotAuthorized, self._allow_sys_no_pm, self.context)
+        self.context.user = self.testsys
+        self.assertTrue(self._allow_sys_no_pm(self.context))
+
+    @rbac.allow('all')
+    def _allow_all(self, context):
+        return True
+
+    @rbac.allow('none')
+    def _allow_none(self, context):
+        return True
+
+    @rbac.allow('projectmanager')
+    def _allow_project_manager(self, context):
+        return True
+
+    @rbac.allow('sysadmin', 'netadmin')
+    def _allow_sys_and_net(self, context):
+        return True
+
+    @rbac.allow('sysadmin')
+    @rbac.deny('projectmanager')
+    def _allow_sys_no_pm(self, context):
+        return True
+
+if __name__ == "__main__":
+    # TODO: Implement use_fake as an option
+    unittest.main()
diff --git a/nova/tests/users_unittest.py b/nova/tests/users_unittest.py
index ff34b8957..a226de455 100644
--- a/nova/tests/users_unittest.py
+++ b/nova/tests/users_unittest.py
@@ -99,6 +99,16 @@ class UserTestCase(test.BaseTestCase):
         users = self.users.get_users()
         self.assertTrue(filter(lambda u: u.id == 'test1', users))
 
+    def test_101_can_add_user_role(self):
+        self.assertFalse(self.users.has_role('test1', 'itsec'))
+        self.users.add_role('test1', 'itsec')
+        self.assertTrue(self.users.has_role('test1', 'itsec'))
+
+    def test_199_can_remove_user_role(self):
+        self.assertTrue(self.users.has_role('test1', 'itsec'))
+        self.users.remove_role('test1', 'itsec')
+        self.assertFalse(self.users.has_role('test1', 'itsec'))
+
     def test_201_can_create_project(self):
         project = self.users.create_project('testproj', 'test1', 'A test project', ['test1'])
         self.assertTrue(filter(lambda p: p.name == 'testproj', self.users.get_projects()))
@@ -151,6 +161,22 @@ class UserTestCase(test.BaseTestCase):
         else:
             self.assertFalse(signed_cert.verify(cloud_cert.get_pubkey()))
 
+    def test_210_can_add_project_role(self):
+        project = self.users.get_project('testproj')
+        self.assertFalse(project.has_role('test1', 'sysadmin'))
+        self.users.add_role('test1', 'sysadmin')
+        self.assertFalse(project.has_role('test1', 'sysadmin'))
+        project.add_role('test1', 'sysadmin')
+        self.assertTrue(project.has_role('test1', 'sysadmin'))
+
+    def test_211_can_remove_project_role(self):
+        project = self.users.get_project('testproj')
+        self.assertTrue(project.has_role('test1', 'sysadmin'))
+        project.remove_role('test1', 'sysadmin')
+        self.assertFalse(project.has_role('test1', 'sysadmin'))
+        self.users.remove_role('test1', 'sysadmin')
+        self.assertFalse(project.has_role('test1', 'sysadmin'))
+
     def test_299_can_delete_project(self):
         self.users.delete_project('testproj')
         self.assertFalse(filter(lambda p: p.name == 'testproj', self.users.get_projects()))
diff --git a/run_tests.py b/run_tests.py
index f80f0af16..847911032 100644
--- a/run_tests.py
+++ b/run_tests.py
@@ -44,6 +44,7 @@ from twisted.scripts import trial as trial_script
 from nova import flags
 from nova import twistd
 
+from nova.tests.access_unittest import *
 from nova.tests.api_unittest import *
 from nova.tests.cloud_unittest import *
 from nova.tests.keeper_unittest import *