Introduce cloud admin role with super user previledges. It is configu…

…rable

and set to 'admin' by default for backward compatability. If set to another
value, 'admin' role will no longer be treated in a special way when multi
tenancy with RBAC is turned on.

Change-Id: Id8e19027ccb93d7320da2c54f279a684422305ba
Partial-Bug: #1530218

src/config/api-server/tests/test_crud_basic.py

@@ -2080,7 +2080,7 @@ def fake_static_file(*args, **kwargs):
             self.assertThat(resp.status_code, Equals(200))
 
         logger.info("Negative case without Documentation")
-        url = 'http://%s:%s/' %(listen_ip, listen_port)
+        url = 'http://%s:%s/virtual-networks' %(listen_ip, listen_port)
         orig_rbac_role = TestLocalAuth._rbac_role
         try:
             TestLocalAuth._rbac_role = 'foobar'

src/config/api-server/tests/test_perms2.py

@@ -79,13 +79,18 @@ def __init__(self, apis_ip, apis_port, kc, name, password, role, project):
            if tenant.name == self.project:
                 break
        self.project_uuid = tenant.id
+       self.tenant = tenant
 
        if self.name not in kc_users:
            logger.info( 'user %s missing from keystone ... creating' % self.name)
-           user = kc.users.create(self.name, self.password, '', tenant_id=tenant.id)
+           kc.users.create(self.name, self.password, '', tenant_id=tenant.id)
 
        role_dict = {role.name:role for role in kc.roles.list()}
        user_dict = {user.name:user for user in kc.users.list()}
+       self.user = user_dict[self.name]
+
+       # update tenant ID (needed if user entry already existed in keystone)
+       self.user.tenant_id = tenant.id
 
        logger.info( 'Adding user %s with role %s to tenant %s' \
             % (name, role, project))
@@ -312,7 +317,7 @@ def get_token(self):
 # This is needed for VncApi._authenticate invocation from within Api server.
 # We don't have access to user information so we hard code admin credentials.
 def ks_admin_authenticate(self, response=None, headers=None):
-    rval = token_from_user_info('admin', 'admin', 'default-domain', 'admin')
+    rval = token_from_user_info('admin', 'admin', 'default-domain', 'cloud-admin')
     new_headers = {}
     new_headers['X-AUTH-TOKEN'] = rval
     return new_headers
@@ -332,6 +337,7 @@ def setUpClass(cls):
                             test_utils.FakeAuthProtocol)]
         extra_config_knobs = [
             ('DEFAULTS', 'multi_tenancy_with_rbac', 'True'),
+            ('DEFAULTS', 'cloud_admin_role', 'cloud-admin'),
             ('DEFAULTS', 'auth', 'keystone'),
         ]
         super(TestPermissions, cls).setUpClass(extra_mocks=extra_mocks,
@@ -347,41 +353,40 @@ def setUp(self):
                        auth_url='http://127.0.0.1:5000/v2.0')
 
         # prepare token before vnc api invokes keystone
-        alice = User(ip, port, kc, 'alice', 'alice123', 'alice-role', 'alice-proj-%s' % self.id())
-        bob =   User(ip, port, kc, 'bob', 'bob123', 'bob-role', 'bob-proj-%s' % self.id())
-        admin = User(ip, port, kc, 'admin', 'contrail123', 'admin', 'admin-%s' % self.id())
+        self.alice = User(ip, port, kc, 'alice', 'alice123', 'alice-role', 'alice-proj-%s' % self.id())
+        self.bob =   User(ip, port, kc, 'bob', 'bob123', 'bob-role', 'bob-proj-%s' % self.id())
+        self.admin = User(ip, port, kc, 'admin', 'contrail123', 'cloud-admin', 'admin-%s' % self.id())
+        self.admin1 = User(ip, port, kc, 'admin1', 'contrail123', 'admin', 'admin1-%s' % self.id())
+        self.admin2 = User(ip, port, kc, 'admin2', 'contrail123', 'admin', 'admin2-%s' % self.id())
 
-        self.alice = alice
-        self.bob   = bob
-        self.admin = admin
-        self.users = [self.alice, self.bob]
+        self.users = [self.alice, self.bob, self.admin1, self.admin2]
 
         """
         1. create project in API server
         2. read objects back and pupolate locally
         3. reassign ownership of projects to user from admin
         """
-        for user in [admin, alice, bob]:
+        for user in [self.admin, self.alice, self.bob, self.admin1, self.admin2]:
             project_obj = Project(user.project)
             project_obj.uuid = user.project_uuid
             logger.info( 'Creating Project object for %s, uuid %s' \
                 % (user.project, user.project_uuid))
-            admin.vnc_lib.project_create(project_obj)
+            self.admin.vnc_lib.project_create(project_obj)
 
             # read projects back
-            user.project_obj = vnc_read_obj(admin.vnc_lib,
+            user.project_obj = vnc_read_obj(self.admin.vnc_lib,
                 'project', obj_uuid = user.project_uuid)
 
             logger.info( 'Change owner of project %s to %s' % (user.project, user.project_uuid))
             set_perms(user.project_obj, owner=user.project_uuid, share = [])
-            admin.vnc_lib.project_update(user.project_obj)
+            self.admin.vnc_lib.project_update(user.project_obj)
 
         # delete test VN if it exists
-        vn_fq_name = [self.domain_name, alice.project, self.vn_name]
-        vn = vnc_read_obj(admin.vnc_lib, 'virtual-network', name = vn_fq_name)
+        vn_fq_name = [self.domain_name, self.alice.project, self.vn_name]
+        vn = vnc_read_obj(self.admin.vnc_lib, 'virtual-network', name = vn_fq_name)
         if vn:
             logger.info( '%s exists ... deleting to start fresh' % vn_fq_name)
-            admin.vnc_lib.virtual_network_delete(fq_name = vn_fq_name)
+            self.admin.vnc_lib.virtual_network_delete(fq_name = vn_fq_name)
 
         # allow permission to create objects
         for user in self.users:
@@ -390,6 +395,69 @@ def setUp(self):
             vnc_fix_api_access_list(self.admin.vnc_lib, user.project_obj,
                 rule_str = '* %s:CRUD' % user.role)
 
+    def test_delete_non_admin_role(self):
+        alice = self.alice
+        bob   = self.bob
+        admin = self.admin
+
+        # allow permission to create all objects
+        for user in self.users:
+            logger.info( "%s: project %s to allow full access to role %s" % \
+                (user.name, user.project, user.role))
+            vnc_fix_api_access_list(self.admin.vnc_lib, user.project_obj,
+                rule_str = '* %s:CRUD' % user.role)
+
+        vn_fq_name = [self.domain_name, alice.project, self.vn_name]
+
+        # delete test VN if it exists
+        if vnc_read_obj(admin.vnc_lib, 'virtual-network', name = vn_fq_name):
+            logger.info( '%s exists ... deleting to start fresh' % vn_fq_name)
+            admin.vnc_lib.virtual_network_delete(fq_name = vn_fq_name)
+
+        vn = VirtualNetwork(self.vn_name, self.alice.project_obj)
+        self.alice.vnc_lib.virtual_network_create(vn)
+
+        # bob - delete VN  ... should fail
+        try:
+            bob.vnc_lib.virtual_network_delete(fq_name = vn_fq_name)
+            self.assertTrue(False, '*** Bob Deleted VN ... test failed!')
+        except PermissionDenied as e:
+            self.assertTrue(True, 'Error deleting VN ... test passed!')
+
+        # Alice - delete VN  ... should succeed
+        try:
+            alice.vnc_lib.virtual_network_delete(fq_name = vn_fq_name)
+            self.assertTrue(True, 'Deleted VN ... test succeeded!')
+        except PermissionDenied as e:
+            self.assertTrue(False, '*** Alice Error deleting VN ... test failed!')
+    # end
+
+    def test_delete_admin_role(self):
+        vn = VirtualNetwork('admin1-vn', self.admin1.project_obj)
+        vn_fq_name = vn.get_fq_name()
+
+        # delete test VN if it exists
+        if vnc_read_obj(self.admin.vnc_lib, 'virtual-network', name = vn_fq_name):
+            logger.info( '%s exists ... deleting to start fresh' % vn_fq_name)
+            self.admin.vnc_lib.virtual_network_delete(fq_name = vn_fq_name)
+
+        self.admin1.vnc_lib.virtual_network_create(vn)
+
+        # admin2 - delete VN  ... should fail
+        try:
+            self.admin2.vnc_lib.virtual_network_delete(fq_name = vn_fq_name)
+            self.assertTrue(False, '*** Deleted VN ... test failed!')
+        except PermissionDenied as e:
+            self.assertTrue(True, 'Error deleting VN ... test passed!')
+
+        # admin1 - delete VN  ... should succeed
+        try:
+            self.admin1.vnc_lib.virtual_network_delete(fq_name = vn_fq_name)
+            self.assertTrue(True, 'Deleted VN ... test succeeded!')
+        except PermissionDenied as e:
+            self.assertTrue(False, '*** Error deleting VN ... test failed!')
+    # end
+
     # delete api-access-list for alice and bob and disallow api access to their projects
     # then try to create VN in the project. This should fail
     def test_api_access(self):
@@ -680,7 +748,7 @@ def test_check_obj_perms_api(self):
         self.assertThat(testfail, Equals(False))
 
         ExpectedPerms = {'admin':'RWX', 'alice':'RWX', 'bob':''}
-        for user in self.users:
+        for user in [alice, bob, admin]:
             perms = user.check_perms(vn.get_uuid())
             self.assertEquals(perms, ExpectedPerms[user.name])
 
@@ -691,7 +759,7 @@ def test_check_obj_perms_api(self):
         alice.vnc_lib.virtual_network_update(vn)
 
         ExpectedPerms = {'admin':'RWX', 'alice':'RWX', 'bob':'R'}
-        for user in self.users:
+        for user in [alice, bob, admin]:
             perms = user.check_perms(vn.get_uuid())
             self.assertEquals(perms, ExpectedPerms[user.name])
 
@@ -702,7 +770,7 @@ def test_check_obj_perms_api(self):
         alice.vnc_lib.virtual_network_update(vn)
 
         ExpectedPerms = {'admin':'RWX', 'alice':'RWX', 'bob':''}
-        for user in self.users:
+        for user in [alice, bob, admin]:
             perms = user.check_perms(vn.get_uuid())
             self.assertEquals(perms, ExpectedPerms[user.name])
         logger.info( 'Reading VN as bob ... should fail')
@@ -714,7 +782,7 @@ def test_check_obj_perms_api(self):
         alice.vnc_lib.virtual_network_update(vn)
 
         ExpectedPerms = {'admin':'RWX', 'alice':'RWX', 'bob':'R'}
-        for user in self.users:
+        for user in [alice, bob, admin]:
             perms = user.check_perms(vn.get_uuid())
             self.assertEquals(perms, ExpectedPerms[user.name])
         logger.info( 'Reading VN as bob ... should fail')
@@ -726,7 +794,7 @@ def test_check_obj_perms_api(self):
         alice.vnc_lib.virtual_network_update(vn)
 
         ExpectedPerms = {'admin':'RWX', 'alice':'RWX', 'bob':'RW'}
-        for user in self.users:
+        for user in [alice, bob, admin]:
             perms = user.check_perms(vn.get_uuid())
             self.assertEquals(perms, ExpectedPerms[user.name])
 

src/config/api-server/utils.py

@@ -15,6 +15,7 @@
 _WEB_HOST = '0.0.0.0'
 _WEB_PORT = 8082
 _ADMIN_PORT = 8095
+_CLOUD_ADMIN_ROLE = 'admin'
 
 def parse_args(args_str):
     args_obj = None
@@ -52,7 +53,7 @@ def parse_args(args_str):
         'logging_conf': '',
         'logger_class': None,
         'multi_tenancy': True,
-        'multi_tenancy_with_rbac': False,
+        'multi_tenancy_with_rbac': True,
         'disc_server_ip': None,
         'disc_server_port': '5998',
         'zk_server_ip': '127.0.0.1:2181',
@@ -69,6 +70,7 @@ def parse_args(args_str):
         'sandesh_send_rate_limit': SandeshSystem.get_sandesh_send_rate_limit(),
         'ifmap_health_check_interval': '60', # in seconds
         'stale_lock_seconds': '5', # lock but no resource past this => stale
+        'cloud_admin_role': _CLOUD_ADMIN_ROLE,
     }
     # ssl options
     secopts = {
@@ -282,6 +284,8 @@ def parse_args(args_str):
             help="Interval seconds to check for ifmap health, default 60")
     parser.add_argument("--stale_lock_seconds",
             help="Time after which lock without resource is stale, default 60")
+    parser.add_argument( "--cloud_admin_role",
+        help="Role name of cloud administrator")
     args_obj, remaining_argv = parser.parse_known_args(remaining_argv)
     args_obj.config_sections = config
     if type(args_obj.cassandra_server_list) is str:

src/config/api-server/vnc_auth_keystone.py

@@ -93,11 +93,7 @@ def set_mt(self, value):
         self.mt = value
 
     def __call__(self, env, start_response):
-        if (env.get('PATH_INFO') and
-            env['PATH_INFO'].startswith('/documentation')):
-            app = bottle.app()
-        else:
-            app = self.app if self.mt else bottle.app()
+        app = self.app if self.mt else bottle.app()
 
         return app(env, start_response)