Description:

1731055
Exceptions on CUD operations because of RBAC permissions
are not handled correctly in VNC openstack.

1730024
Perms2 is a complex property. PUT operation when issued by user
may carry partial attributes of that object. So direct dereferencing
is incorrect.
If that attribute is not present, existing attribute stays.
If that attribue needs to be deleted, it has to be present.

Change-Id: I4b3abdd24c57e7da49d728f2073ed9c2a3828cb0
Closes-Bug: #1731055
Closes-Bug: #1730024
(cherry picked from commit debdfb6)

src/config/api-server/vnc_db.py

@@ -1030,8 +1030,8 @@ def wrapper2(self, obj_type, obj_id, obj_dict):
                 if not new_perms2:
                     return (ok, result)
 
-                share_perms = new_perms2['share']
-                global_access = new_perms2['global_access']
+                share_perms = new_perms2.get('share', cur_perms2['share'])
+                global_access = new_perms2.get('global_access', cur_perms2['global_access'])
 
                 # msg = 'RBAC: BSL perms new %s, cur %s' % (new_perms2, cur_perms2)
                 # self.config_log(msg, level=SandeshLevel.SYS_NOTICE)
@@ -1045,7 +1045,7 @@ def wrapper2(self, obj_type, obj_id, obj_dict):
 
                 # change in shared list? Construct temporary sets to compare
                 cur_shared_list = set(item['tenant']+':'+str(item['tenant_access']) for item in cur_perms2['share'])
-                new_shared_list = set(item['tenant']+':'+str(item['tenant_access']) for item in new_perms2['share'])
+                new_shared_list = set(item['tenant']+':'+str(item['tenant_access']) for item in share_perms)
                 if cur_shared_list == new_shared_list:
                     return (ok, result)
 

src/config/vnc_openstack/vnc_openstack/neutron_plugin_db.py

@@ -165,6 +165,11 @@ def _request(self, op, url, data=None, *args, **kwargs):
                 pass
 
             return ret_val
+
+        except PermissionDenied as e:
+            exc_info = {'exception': 'NotAuthorized'}
+            exc_info.update(msg=str(e))
+            bottle.abort(400, json.dumps(exc_info))
         finally:
             if had_user_token:
                 self._headers['X-AUTH-TOKEN'] = orig_user_token
@@ -361,7 +366,7 @@ def _security_group_rule_create(self, sg_id, sg_rule):
 
         try:
             self._resource_update('security_group', sg_vnc)
-        except (BadRequest, PermissionDenied) as e:
+        except BadRequest as e:
             self._raise_contrail_exception('BadRequest',
                 resource='security_group_rule', msg=str(e))
         except OverQuota as e:
@@ -441,7 +446,7 @@ def _resource_create(self, resource_type, obj):
                 obj.name += '-' + obj.uuid
                 obj.fq_name[-1] += '-' + obj.uuid
                 obj_uuid = create_method(obj)
-        except (PermissionDenied, BadRequest) as e:
+        except BadRequest as e:
             self._raise_contrail_exception('BadRequest',
                 resource=resource_type, msg=str(e))
         except OverQuota as e:
@@ -480,7 +485,7 @@ def _virtual_network_read(self, net_id=None, fq_name=None, fields=None):
     def _virtual_network_update(self, net_obj):
         try:
             self._resource_update('virtual_network', net_obj)
-        except (PermissionDenied, RefsExistError) as e:
+        except RefsExistError as e:
             self._raise_contrail_exception('BadRequest',
                 resource='network', msg=str(e))
     # end _virtual_network_update

src/config/vnc_openstack/vnc_openstack/tests/test_basic.py

@@ -1707,6 +1707,68 @@ def test_post_neutron_checks_auth_token(self):
     # end test_post_neutron_checks_auth_token
 # end class TestAuthenticatedAccess
 
+class TestRBACPerms(test_case.NeutronBackendTestCase):
+    domain_name = 'default-domain'
+    fqdn = [domain_name]
+    _api_session = requests.Session()
+    adapter = requests.adapters.HTTPAdapter()
+    _api_session.mount("http://", adapter)
+
+    @classmethod
+    def setUpClass(cls):
+        from keystonemiddleware import auth_token
+        class FakeAuthProtocol(object):
+            _test_cls = cls
+            def __init__(self, app, *args, **kwargs):
+                self._app = app
+
+            # end __init__
+            def __call__(self, env, start_response):
+                #Count number of calls made
+                if env.get('HTTP_X_AUTH_TOKEN') == 'test123':
+                    env['HTTP_X_ROLE'] = 'test'
+                else:
+                    env['HTTP_X_ROLE'] = 'cloud-admin'
+                return self._app(env, start_response)
+            # end __call__
+        # end class FakeAuthProtocol
+
+        extra_mocks = [(auth_token, 'AuthProtocol',
+                            FakeAuthProtocol)]
+        extra_config_knobs = [
+            ('DEFAULTS', 'aaa_mode', 'rbac'),
+            ('DEFAULTS', 'cloud_admin_role', 'cloud-admin'),
+            ('DEFAULTS', 'global_read_only_role', 'read-only-role'),
+            ('DEFAULTS', 'auth', 'keystone'),
+        ]
+        super(TestRBACPerms, cls).setUpClass(extra_mocks=extra_mocks,
+            extra_config_knobs=extra_config_knobs)
+
+    def test_neutron_perms(self):
+        test_obj = self._create_test_object()
+        proj_obj = self._vnc_lib.project_read(
+            fq_name=['default-domain', 'default-project'])
+        context = {'operation': 'CREATE',
+                   'user_id': '',
+                   'is_admin': False,
+                   'roles': '',
+                   'tenant_id': proj_obj.uuid}
+
+        data = {'resource': {'name': 'test_network',
+                             'tenant_id': proj_obj.uuid}}
+        body = {'context': context, 'data': data}
+        uri = '/neutron/network'
+        url = "http://%s:%s%s" \
+              % (self._vnc_lib._web_host, self._vnc_lib._web_port, uri)
+        headers=self._vnc_lib._headers
+        headers['X_AUTH_TOKEN'] = 'test123'
+        header = json.dumps(headers)
+        body = json.dumps(body)
+        test_pass = False
+        val = TestRBACPerms._api_session.post(url, data=body, headers=headers, verify=False)
+        self.assertIn('NotAuthorized', val._content)
+# end class TestRBACPerms 
+
 class TestKeystoneCallCount(test_case.NeutronBackendTestCase):
     test_obj_uuid = None
     test_failures = []