Merge "Return 403 instead of 404 on attr policy failures"
This commit is contained in:
		@@ -26,6 +26,7 @@ from neutron.api.v2 import resource as wsgi_resource
 | 
			
		||||
from neutron.common import constants as const
 | 
			
		||||
from neutron.common import exceptions
 | 
			
		||||
from neutron.common import rpc as n_rpc
 | 
			
		||||
from neutron.openstack.common import excutils
 | 
			
		||||
from neutron.openstack.common import log as logging
 | 
			
		||||
from neutron import policy
 | 
			
		||||
from neutron import quota
 | 
			
		||||
@@ -517,8 +518,12 @@ class Controller(object):
 | 
			
		||||
                           action,
 | 
			
		||||
                           orig_obj)
 | 
			
		||||
        except exceptions.PolicyNotAuthorized:
 | 
			
		||||
            # To avoid giving away information, pretend that it
 | 
			
		||||
            # doesn't exist
 | 
			
		||||
            with excutils.save_and_reraise_exception() as ctxt:
 | 
			
		||||
                # If a tenant is modifying it's own object, it's safe to return
 | 
			
		||||
                # a 403. Otherwise, pretend that it doesn't exist to avoid
 | 
			
		||||
                # giving away information.
 | 
			
		||||
                if request.context.tenant_id != orig_obj['tenant_id']:
 | 
			
		||||
                    ctxt.reraise = False
 | 
			
		||||
            msg = _('The resource could not be found.')
 | 
			
		||||
            raise webob.exc.HTTPNotFound(msg)
 | 
			
		||||
 | 
			
		||||
 
 | 
			
		||||
@@ -151,14 +151,11 @@ class PortBindingsTestCase(test_db_plugin.NeutronDbPluginV2TestCase):
 | 
			
		||||
            with self.subnet(network=net1) as subnet1:
 | 
			
		||||
                with self.port(subnet=subnet1) as port:
 | 
			
		||||
                    # By default user is admin - now test non admin user
 | 
			
		||||
                    # Note that 404 is returned when prohibit by policy.
 | 
			
		||||
                    # See comment for PolicyNotAuthorized except clause
 | 
			
		||||
                    # in update() in neutron.api.v2.base.Controller.
 | 
			
		||||
                    port_id = port['port']['id']
 | 
			
		||||
                    ctx = self._get_non_admin_context()
 | 
			
		||||
                    port = self._update('ports', port_id,
 | 
			
		||||
                                        {'port': profile_arg},
 | 
			
		||||
                                        expected_code=404,
 | 
			
		||||
                                        expected_code=exc.HTTPForbidden.code,
 | 
			
		||||
                                        neutron_context=ctx)
 | 
			
		||||
 | 
			
		||||
 | 
			
		||||
 
 | 
			
		||||
@@ -103,7 +103,7 @@ class TestNecPortBindingPortInfo(test_nec_plugin.NecPluginV2TestCase):
 | 
			
		||||
            # port-update with non admin user should fail
 | 
			
		||||
            self._update('ports', port_id,
 | 
			
		||||
                         {'port': profile_arg},
 | 
			
		||||
                         expected_code=404,
 | 
			
		||||
                         expected_code=exc.HTTPForbidden.code,
 | 
			
		||||
                         neutron_context=ctx)
 | 
			
		||||
 | 
			
		||||
    def test_port_update_portinfo(self):
 | 
			
		||||
@@ -255,14 +255,11 @@ class TestNecPortBindingPortInfo(test_nec_plugin.NecPluginV2TestCase):
 | 
			
		||||
            with self.subnet(network=net1) as subnet1:
 | 
			
		||||
                with self.port(subnet=subnet1) as port:
 | 
			
		||||
                    # By default user is admin - now test non admin user
 | 
			
		||||
                    # Note that 404 is returned when prohibit by policy.
 | 
			
		||||
                    # See comment for PolicyNotAuthorized except clause
 | 
			
		||||
                    # in update() in neutron.api.v2.base.Controller.
 | 
			
		||||
                    port_id = port['port']['id']
 | 
			
		||||
                    ctx = self._get_non_admin_context()
 | 
			
		||||
                    port = self._update('ports', port_id,
 | 
			
		||||
                                        {'port': profile_arg},
 | 
			
		||||
                                        expected_code=404,
 | 
			
		||||
                                        expected_code=exc.HTTPForbidden.code,
 | 
			
		||||
                                        neutron_context=ctx)
 | 
			
		||||
                self.assertEqual(self.ofc.create_ofc_port.call_count, 0)
 | 
			
		||||
 | 
			
		||||
 
 | 
			
		||||
@@ -279,8 +279,7 @@ class TestFirewallPluginBase(test_db_firewall.TestFirewallDBPlugin):
 | 
			
		||||
                    'firewalls', data, fw_id,
 | 
			
		||||
                    context=context.Context('', 'noadmin'))
 | 
			
		||||
                res = req.get_response(self.ext_api)
 | 
			
		||||
                # returns 404 due to security reasons
 | 
			
		||||
                self.assertEqual(res.status_int, exc.HTTPNotFound.code)
 | 
			
		||||
                self.assertEqual(res.status_int, exc.HTTPForbidden.code)
 | 
			
		||||
 | 
			
		||||
    def test_update_firewall_policy_fails_when_firewall_pending(self):
 | 
			
		||||
        name = "new_firewall1"
 | 
			
		||||
 
 | 
			
		||||
@@ -1805,7 +1805,7 @@ class TestNetworksV2(NeutronDbPluginV2TestCase):
 | 
			
		||||
            res = self.deserialize(self.fmt, req.get_response(self.api))
 | 
			
		||||
            self.assertTrue(res['network']['shared'])
 | 
			
		||||
 | 
			
		||||
    def test_update_network_set_shared_owner_returns_404(self):
 | 
			
		||||
    def test_update_network_set_shared_owner_returns_403(self):
 | 
			
		||||
        with self.network(shared=False) as network:
 | 
			
		||||
            net_owner = network['network']['tenant_id']
 | 
			
		||||
            data = {'network': {'shared': True}}
 | 
			
		||||
@@ -1814,7 +1814,7 @@ class TestNetworksV2(NeutronDbPluginV2TestCase):
 | 
			
		||||
                                          network['network']['id'])
 | 
			
		||||
            req.environ['neutron.context'] = context.Context('u', net_owner)
 | 
			
		||||
            res = req.get_response(self.api)
 | 
			
		||||
            self.assertEqual(res.status_int, webob.exc.HTTPNotFound.code)
 | 
			
		||||
            self.assertEqual(res.status_int, webob.exc.HTTPForbidden.code)
 | 
			
		||||
 | 
			
		||||
    def test_update_network_with_subnet_set_shared(self):
 | 
			
		||||
        with self.network(shared=False) as network:
 | 
			
		||||
 
 | 
			
		||||
@@ -118,8 +118,7 @@ class ExtNetDBTestCase(test_db_plugin.NeutronDbPluginV2TestCase):
 | 
			
		||||
                                          network['network']['id'])
 | 
			
		||||
            req.environ['neutron.context'] = context.Context('', 'noadmin')
 | 
			
		||||
            res = req.get_response(self.api)
 | 
			
		||||
            # The API layer always returns 404 on updates in place of 403
 | 
			
		||||
            self.assertEqual(exc.HTTPNotFound.code, res.status_int)
 | 
			
		||||
            self.assertEqual(exc.HTTPForbidden.code, res.status_int)
 | 
			
		||||
 | 
			
		||||
    def test_network_filter_hook_admin_context(self):
 | 
			
		||||
        plugin = manager.NeutronManager.get_plugin()
 | 
			
		||||
 
 | 
			
		||||
@@ -152,8 +152,8 @@ class ProvidernetExtensionTestCase(testlib_api.WebTestCase):
 | 
			
		||||
        res, _1 = self._post_network_with_provider_attrs(ctx, True)
 | 
			
		||||
        self.assertEqual(res.status_int, web_exc.HTTPForbidden.code)
 | 
			
		||||
 | 
			
		||||
    def test_network_update_with_provider_attrs_noadmin_returns_404(self):
 | 
			
		||||
    def test_network_update_with_provider_attrs_noadmin_returns_403(self):
 | 
			
		||||
        tenant_id = 'no_admin'
 | 
			
		||||
        ctx = context.Context('', tenant_id, is_admin=False)
 | 
			
		||||
        res, _1, _2 = self._put_network_with_provider_attrs(ctx, True)
 | 
			
		||||
        self.assertEqual(res.status_int, web_exc.HTTPNotFound.code)
 | 
			
		||||
        self.assertEqual(res.status_int, web_exc.HTTPForbidden.code)
 | 
			
		||||
 
 | 
			
		||||
@@ -13,6 +13,7 @@
 | 
			
		||||
# See the License for the specific language governing permissions and
 | 
			
		||||
# limitations under the License.
 | 
			
		||||
 | 
			
		||||
from webob import exc
 | 
			
		||||
 | 
			
		||||
from neutron.api.v2 import attributes as attr
 | 
			
		||||
from neutron import context
 | 
			
		||||
@@ -384,9 +385,4 @@ class TestPortSecurity(PortSecurityDBTestCase):
 | 
			
		||||
                req.environ['neutron.context'] = context.Context(
 | 
			
		||||
                    '', 'not_network_owner')
 | 
			
		||||
                res = req.get_response(self.api)
 | 
			
		||||
                # TODO(salvatore-orlando): Expected error is 404 because
 | 
			
		||||
                # the current API controller always returns this error
 | 
			
		||||
                # code for any policy check failures on update.
 | 
			
		||||
                # It should be 404 when the caller cannot access the whole
 | 
			
		||||
                # resource, and 403 when it cannot access a single attribute
 | 
			
		||||
                self.assertEqual(res.status_int, 404)
 | 
			
		||||
                self.assertEqual(res.status_int, exc.HTTPForbidden.code)
 | 
			
		||||
 
 | 
			
		||||
		Reference in New Issue
	
	Block a user