openstack
/
glance
Code Issues Proposed changes

Move metadef property policy checks in the API 

Made provision to enforce metadef property related policy
checks in API.

Partially implements: blueprint policy-refactor

Depends-On: https://review.opendev.org/c/openstack/nova/+/688802

Change-Id: I6440462d16006204a5cd46a826dc281fe23ab992
This commit is contained in:
Abhishek Kekane 2021-07-19 18:14:27 +00:00
parent f094b7d2b2
commit f9deee125c
5 changed files with 457 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

130
glance/api/v2/metadef_properties.py
View File

@ -26,6 +26,7 @@ from glance.api.v2 import metadef_namespaces as namespaces
from glance.api.v2.model.metadef_namespace import Namespace
from glance.api.v2.model.metadef_property_type import PropertyType
from glance.api.v2.model.metadef_property_type import PropertyTypes
from glance.api.v2 import policy as api_policy
from glance.common import exception
from glance.common import wsgi
import glance.db
@ -62,10 +63,29 @@ class NamespacePropertiesController(object):
return property_type
def index(self, req, namespace):
ns_repo = self.gateway.get_metadef_namespace_repo(
req.context, authorization_layer=False)
try:
namespace_obj = ns_repo.get(namespace)
except (exception.Forbidden, exception.NotFound):
# NOTE (abhishekk): Returning 404 Not Found as the
# namespace is outside of this user's project
msg = _("Namespace %s not found") % namespace
raise webob.exc.HTTPNotFound(explanation=msg)
try:
# NOTE(abhishekk): This is just a "do you have permission to
# list properties" check. Each property is checked against
# get_metadef_property below.
api_policy.MetadefAPIPolicy(
req.context,
md_resource=namespace_obj,
enforcer=self.policy).get_metadef_properties()
filters = dict()
filters['namespace'] = namespace
prop_repo = self.gateway.get_metadef_property_repo(req.context)
prop_repo = self.gateway.get_metadef_property_repo(
req.context, authorization_layer=False)
db_properties = prop_repo.list(filters=filters)
property_list = Namespace.to_model_properties(db_properties)
namespace_properties = PropertyTypes()
@ -76,16 +96,35 @@ class NamespacePropertiesController(object):
raise webob.exc.HTTPForbidden(explanation=e.msg)
except exception.NotFound as e:
raise webob.exc.HTTPNotFound(explanation=e.msg)
except Exception as e:
LOG.error(encodeutils.exception_to_unicode(e))
raise webob.exc.HTTPInternalServerError()
return namespace_properties
def show(self, req, namespace, property_name, filters=None):
ns_repo = self.gateway.get_metadef_namespace_repo(
req.context, authorization_layer=False)
try:
namespace_obj = ns_repo.get(namespace)
except (exception.Forbidden, exception.NotFound):
# NOTE (abhishekk): Returning 404 Not Found as the
# namespace is outside of this user's project
msg = _("Namespace %s not found") % namespace
raise webob.exc.HTTPNotFound(explanation=msg)
try:
# NOTE(abhishekk): Metadef properties are associated with
# namespace, so made provision to pass namespace here
# for visibility check
api_pol = api_policy.MetadefAPIPolicy(
req.context,
md_resource=namespace_obj,
enforcer=self.policy)
api_pol.get_metadef_property()
if filters and filters['resource_type']:
# Verify that you can fetch resource type details
api_pol.get_metadef_resource_type()
rs_repo = self.gateway.get_metadef_resource_type_repo(
req.context)
req.context, authorization_layer=False)
db_resource_type = rs_repo.get(filters['resource_type'],
namespace)
prefix = db_resource_type.prefix
@ -99,7 +138,8 @@ class NamespacePropertiesController(object):
'prefix': prefix})
raise exception.NotFound(msg)
prop_repo = self.gateway.get_metadef_property_repo(req.context)
prop_repo = self.gateway.get_metadef_property_repo(
req.context, authorization_layer=False)
db_property = prop_repo.get(namespace, property_name)
property = self._to_model(db_property)
except exception.Forbidden as e:
@ -108,15 +148,32 @@ class NamespacePropertiesController(object):
raise webob.exc.HTTPForbidden(explanation=e.msg)
except exception.NotFound as e:
raise webob.exc.HTTPNotFound(explanation=e.msg)
except Exception as e:
LOG.error(encodeutils.exception_to_unicode(e))
raise webob.exc.HTTPInternalServerError()
return property
def create(self, req, namespace, property_type):
prop_factory = self.gateway.get_metadef_property_factory(req.context)
prop_repo = self.gateway.get_metadef_property_repo(req.context)
prop_factory = self.gateway.get_metadef_property_factory(
req.context, authorization_layer=False)
prop_repo = self.gateway.get_metadef_property_repo(
req.context, authorization_layer=False)
ns_repo = self.gateway.get_metadef_namespace_repo(
req.context, authorization_layer=False)
try:
namespace_obj = ns_repo.get(namespace)
except (exception.Forbidden, exception.NotFound):
# NOTE (abhishekk): Returning 404 Not Found as the
# namespace is outside of this user's project
msg = _("Namespace %s not found") % namespace
raise webob.exc.HTTPNotFound(explanation=msg)
try:
# NOTE(abhishekk): Metadef property is created for Metadef
# namespaces. Here we are just checking if user is authorized
# to create metadef property or not.
api_policy.MetadefAPIPolicy(
req.context,
md_resource=namespace_obj,
enforcer=self.policy).add_metadef_property()
new_property_type = prop_factory.new_namespace_property(
namespace=namespace, **self._to_dict(property_type))
prop_repo.add(new_property_type)
@ -132,14 +189,30 @@ class NamespacePropertiesController(object):
raise webob.exc.HTTPNotFound(explanation=e.msg)
except exception.Duplicate as e:
raise webob.exc.HTTPConflict(explanation=e.msg)
except Exception as e:
LOG.error(encodeutils.exception_to_unicode(e))
raise webob.exc.HTTPInternalServerError()
return self._to_model(new_property_type)
def update(self, req, namespace, property_name, property_type):
prop_repo = self.gateway.get_metadef_property_repo(req.context)
prop_repo = self.gateway.get_metadef_property_repo(
req.context, authorization_layer=False)
ns_repo = self.gateway.get_metadef_namespace_repo(
req.context, authorization_layer=False)
try:
namespace_obj = ns_repo.get(namespace)
except (exception.Forbidden, exception.NotFound):
# NOTE (abhishekk): Returning 404 Not Found as the
# namespace is outside of this user's project
msg = _("Namespace %s not found") % namespace
raise webob.exc.HTTPNotFound(explanation=msg)
try:
# NOTE(abhishekk): Metadef property is created for Metadef
# namespaces. Here we are just checking if user is authorized
# to update metadef property or not.
api_policy.MetadefAPIPolicy(
req.context,
md_resource=namespace_obj,
enforcer=self.policy).modify_metadef_property()
db_property_type = prop_repo.get(namespace, property_name)
db_property_type._old_name = db_property_type.name
db_property_type.name = property_type.name
@ -157,14 +230,30 @@ class NamespacePropertiesController(object):
raise webob.exc.HTTPNotFound(explanation=e.msg)
except exception.Duplicate as e:
raise webob.exc.HTTPConflict(explanation=e.msg)
except Exception as e:
LOG.error(encodeutils.exception_to_unicode(e))
raise webob.exc.HTTPInternalServerError()
return self._to_model(updated_property_type)
def delete(self, req, namespace, property_name):
prop_repo = self.gateway.get_metadef_property_repo(req.context)
prop_repo = self.gateway.get_metadef_property_repo(
req.context, authorization_layer=False)
ns_repo = self.gateway.get_metadef_namespace_repo(
req.context, authorization_layer=False)
try:
namespace_obj = ns_repo.get(namespace)
except (exception.Forbidden, exception.NotFound):
# NOTE (abhishekk): Returning 404 Not Found as the
# namespace is outside of this user's project
msg = _("Namespace %s not found") % namespace
raise webob.exc.HTTPNotFound(explanation=msg)
try:
# NOTE(abhishekk): Metadef property is created for Metadef
# namespaces. Here we are just checking if user is authorized
# to delete metadef property or not.
api_policy.MetadefAPIPolicy(
req.context,
md_resource=namespace_obj,
enforcer=self.policy).remove_metadef_property()
property_type = prop_repo.get(namespace, property_name)
property_type.delete()
prop_repo.remove(property_type)
@ -174,9 +263,6 @@ class NamespacePropertiesController(object):
raise webob.exc.HTTPForbidden(explanation=e.msg)
except exception.NotFound as e:
raise webob.exc.HTTPNotFound(explanation=e.msg)
except Exception as e:
LOG.error(encodeutils.exception_to_unicode(e))
raise webob.exc.HTTPInternalServerError()
class RequestDeserializer(wsgi.JSONRequestDeserializer):

9
glance/api/v2/policy.py
View File

@ -360,6 +360,15 @@ class MetadefAPIPolicy(APIPolicyBase):
def get_metadef_properties(self):
self._enforce('get_metadef_properties')
def remove_metadef_property(self):
self._enforce('remove_metadef_property')
def get_metadef_property(self):
self._enforce('get_metadef_property')
def modify_metadef_property(self):
self._enforce('modify_metadef_property')
def add_metadef_resource_type_association(self):
self._enforce('add_metadef_resource_type_association')

320
glance/tests/functional/v2/test_metadef_property_api_policy.py Normal file
View File

@ -0,0 +1,320 @@
# Copyright 2021 Red Hat, Inc.
# All Rights Reserved.
#
# 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.
from unittest import mock
import oslo_policy.policy
from glance.api import policy
from glance.tests import functional
PROPERTY1 = {
"name": "MyProperty",
"title": "My Property",
"description": "My Property for My Namespace",
"type": "string"
}
PROPERTY2 = {
"name": "MySecondProperty",
"title": "My Second Property",
"description": "My Second Property for My Namespace",
"type": "string"
}
NAME_SPACE1 = {
"namespace": "MyNamespace",
"display_name": "My User Friendly Namespace",
"description": "My description"
}
class TestMetadefPropertiesPolicy(functional.SynchronousAPIBase):
def setUp(self):
super(TestMetadefPropertiesPolicy, self).setUp()
self.policy = policy.Enforcer(suppress_deprecation_warnings=True)
def load_data(self, create_properties=False):
path = '/v2/metadefs/namespaces'
md_resource = self._create_metadef_resource(path=path,
data=NAME_SPACE1)
self.assertEqual('MyNamespace', md_resource['namespace'])
if create_properties:
namespace = md_resource['namespace']
path = '/v2/metadefs/namespaces/%s/properties' % namespace
for prop in [PROPERTY1, PROPERTY2]:
md_resource = self._create_metadef_resource(path=path,
data=prop)
self.assertEqual(prop['name'], md_resource['name'])
def set_policy_rules(self, rules):
self.policy.set_rules(
oslo_policy.policy.Rules.from_dict(rules),
overwrite=True)
def start_server(self):
with mock.patch.object(policy, 'Enforcer') as mock_enf:
mock_enf.return_value = self.policy
super(TestMetadefPropertiesPolicy, self).start_server()
def _verify_forbidden_converted_to_not_found(self, path, method,
json=None):
# Note for other reviewers, these tests runs by default using
# admin role, to test this scenario we need private namespace
# of current project to be accessed by other projects non-admin
# user.
headers = self._headers({
'X-Tenant-Id': 'fake-tenant-id',
'X-Roles': 'member',
})
resp = self.api_request(method, path, headers=headers, json=json)
self.assertEqual(404, resp.status_code)
def test_property_list_basic(self):
self.start_server()
# Create namespace and properties
self.load_data(create_properties=True)
# First make sure list property works with default policy
namespace = NAME_SPACE1['namespace']
path = '/v2/metadefs/namespaces/%s/properties' % namespace
resp = self.api_get(path)
md_resource = resp.json
self.assertEqual(2, len(md_resource['properties']))
# Now disable get_metadef_properties permissions and make sure
# any other attempts fail
self.set_policy_rules({
'get_metadef_properties': '!',
'get_metadef_namespace': ''
})
resp = self.api_get(path)
self.assertEqual(403, resp.status_code)
# Now disable both permissions and make sure you will get
# 404 Not Found
self.set_policy_rules({
'get_metadef_properties': '!',
'get_metadef_namespace': '!'
})
resp = self.api_get(path)
# Note for reviewers, this causes our "check get if list fails"
# logic to return 404 as we expect, but not related to the latest
# rev that checks the namespace get operation first.
self.assertEqual(404, resp.status_code)
# Ensure accessing non visible namespace will catch 403 and
# return 404 to user
self.set_policy_rules({
'get_metadef_properties': '',
'get_metadef_namespace': ''
})
self._verify_forbidden_converted_to_not_found(path, 'GET')
def test_property_get_basic(self):
self.start_server()
# Create namespace and properties
self.load_data(create_properties=True)
# First make sure get property works with default policy
path = '/v2/metadefs/namespaces/%s/properties/%s' % (
NAME_SPACE1['namespace'], PROPERTY1['name'])
resp = self.api_get(path)
md_resource = resp.json
self.assertEqual(PROPERTY1['name'], md_resource['name'])
# Now disable get_metadef_property permissions and make sure any other
# attempts fail
self.set_policy_rules({
'get_metadef_property': '!',
'get_metadef_namespace': '',
'get_metadef_resource_type': ''
})
resp = self.api_get(path)
self.assertEqual(403, resp.status_code)
# Now disable get_metadef_resource_type permissions and make sure
# any other attempts fail
self.set_policy_rules({
'get_metadef_property': '',
'get_metadef_namespace': '',
'get_metadef_resource_type': '!'
})
url_path = "%s?resource_type='abcd'" % path
resp = self.api_get(url_path)
self.assertEqual(403, resp.status_code)
# Now disable all permissions and make sure you will get
# 404 Not Found
self.set_policy_rules({
'get_metadef_property': '!',
'get_metadef_namespace': '!',
'get_metadef_resource_type': '!'
})
resp = self.api_get(path)
# Note for reviewers, this causes our "check get if get fails"
# logic to return 404 as we expect, but not related to the latest
# rev that checks the namespace get operation first.
self.assertEqual(404, resp.status_code)
# Ensure accessing non visible namespace will catch 403 and
# return 404 to user
self.set_policy_rules({
'get_metadef_property': '',
'get_metadef_namespace': '',
'get_metadef_resource_type': ''
})
self._verify_forbidden_converted_to_not_found(path, 'GET')
def test_property_create_basic(self):
self.start_server()
# Create namespace
self.load_data()
# First make sure create property works with default policy
namespace = NAME_SPACE1['namespace']
path = '/v2/metadefs/namespaces/%s/properties' % namespace
md_resource = self._create_metadef_resource(path=path,
data=PROPERTY1)
self.assertEqual('MyProperty', md_resource['name'])
# Now disable add_metadef_property permissions and make sure any other
# attempts fail
self.set_policy_rules({
'add_metadef_property': '!',
'get_metadef_namespace': ''
})
resp = self.api_post(path, json=PROPERTY2)
self.assertEqual(403, resp.status_code)
# Now disable both permissions and make sure you will get
# 404 Not Found
self.set_policy_rules({
'add_metadef_property': '!',
'get_metadef_namespace': '!'
})
resp = self.api_post(path, json=PROPERTY2)
# Note for reviewers, this causes our "check get if get fails"
# logic to return 404 as we expect, but not related to the latest
# rev that checks the namespace get operation first.
self.assertEqual(404, resp.status_code)
# Ensure accessing non visible namespace will catch 403 and
# return 404 to user
self.set_policy_rules({
'add_metadef_property': '',
'get_metadef_namespace': ''
})
self._verify_forbidden_converted_to_not_found(path, 'POST',
json=PROPERTY2)
def test_property_update_basic(self):
self.start_server()
# Create namespace and properties
self.load_data(create_properties=True)
# First make sure update property works with default policy
path = '/v2/metadefs/namespaces/%s/properties/%s' % (
NAME_SPACE1['namespace'], PROPERTY1['name'])
data = {
"name": PROPERTY1['name'],
"title": PROPERTY1['title'],
"type": PROPERTY1['type'],
"description": "My updated description"
}
resp = self.api_put(path, json=data)
md_resource = resp.json
self.assertEqual(data['description'], md_resource['description'])
# Now disable modify_metadef_property permissions and make sure
# any other attempts fail
data = {
"name": PROPERTY2['name'],
"title": PROPERTY2['title'],
"type": PROPERTY2['type'],
"description": "My updated description"
}
path = '/v2/metadefs/namespaces/%s/properties/%s' % (
NAME_SPACE1['namespace'], PROPERTY2['name'])
self.set_policy_rules({
'modify_metadef_property': '!',
'get_metadef_namespace': ''
})
resp = self.api_put(path, json=data)
self.assertEqual(403, resp.status_code)
# Now disable both permissions and make sure you will get
# 404 Not Found
self.set_policy_rules({
'modify_metadef_property': '!',
'get_metadef_namespace': '!'
})
resp = self.api_put(path, json=data)
# Note for reviewers, this causes our "check get if get fails"
# logic to return 404 as we expect, but not related to the latest
# rev that checks the namespace get operation first.
self.assertEqual(404, resp.status_code)
# Ensure accessing non visible namespace will catch 403 and
# return 404 to user
self.set_policy_rules({
'modify_metadef_property': '',
'get_metadef_namespace': ''
})
self._verify_forbidden_converted_to_not_found(path, 'PUT',
json=data)
def test_property_delete_basic(self):
self.start_server()
# Create namespace and properties
self.load_data(create_properties=True)
# Now ensure you are able to delete the property
path = '/v2/metadefs/namespaces/%s/properties/%s' % (
NAME_SPACE1['namespace'], PROPERTY1['name'])
resp = self.api_delete(path)
self.assertEqual(204, resp.status_code)
# Verify that property is deleted
path = "/v2/metadefs/namespaces/%s/properties/%s" % (
NAME_SPACE1['namespace'], PROPERTY1['name'])
resp = self.api_get(path)
self.assertEqual(404, resp.status_code)
# Now disable remove_metadef_property permissions and make sure
# any other attempts fail
path = '/v2/metadefs/namespaces/%s/properties/%s' % (
NAME_SPACE1['namespace'], PROPERTY2['name'])
self.set_policy_rules({
'remove_metadef_property': '!',
'get_metadef_namespace': ''
})
resp = self.api_delete(path)
self.assertEqual(403, resp.status_code)
# Now disable both permissions and make sure you will get
# 404 Not Found
self.set_policy_rules({
'remove_metadef_property': '!',
'get_metadef_namespace': '!'
})
resp = self.api_delete(path)
# Note for reviewers, this causes our "check get if get fails"
# logic to return 404 as we expect, but not related to the latest
# rev that checks the namespace get operation first.
self.assertEqual(404, resp.status_code)
# Ensure accessing non visible namespace will catch 403 and
# return 404 to user
self.set_policy_rules({
'remove_metadef_property': '',
'get_metadef_namespace': ''
})
self._verify_forbidden_converted_to_not_found(path, 'DELETE')

8
glance/tests/unit/v2/test_metadef_resources.py
View File

@ -936,7 +936,7 @@ class TestMetadefsControllers(base.IsolatedUnitTest):
def test_property_show_non_existing_resource_type(self):
request = unit_test_utils.get_fake_request()
self.assertRaises(webob.exc.HTTPForbidden,
self.assertRaises(webob.exc.HTTPNotFound,
self.property_controller.show, request, NAMESPACE2,
PROPERTY1, filters={'resource_type': 'test'})
@ -997,14 +997,14 @@ class TestMetadefsControllers(base.IsolatedUnitTest):
PROPERTY1)
def test_property_delete_non_existing(self):
request = unit_test_utils.get_fake_request()
request = unit_test_utils.get_fake_request(roles=['admin'])
self.assertRaises(webob.exc.HTTPNotFound,
self.property_controller.delete, request, NAMESPACE5,
PROPERTY2)
self.assertNotificationsLog([])
def test_property_delete_non_existing_namespace(self):
request = unit_test_utils.get_fake_request()
request = unit_test_utils.get_fake_request(roles=['admin'])
self.assertRaises(webob.exc.HTTPNotFound,
self.property_controller.delete, request, NAMESPACE4,
PROPERTY1)
@ -1114,7 +1114,7 @@ class TestMetadefsControllers(base.IsolatedUnitTest):
property.type = 'string'
property.title = 'title'
self.assertRaises(webob.exc.HTTPForbidden,
self.assertRaises(webob.exc.HTTPNotFound,
self.property_controller.create, request, NAMESPACE1,
property)
self.assertNotificationsLog([])

16
glance/tests/unit/v2/test_v2_policy.py
View File

@ -579,6 +579,22 @@ class TestMetadefAPIPolicy(APIPolicyBase):
'get_metadef_properties',
mock.ANY)
def test_get_metadef_property(self):
self.policy.get_metadef_property()
self.enforcer.enforce.assert_called_once_with(self.context,
'get_metadef_property',
mock.ANY)
def test_modify_metadef_property(self):
self.policy.modify_metadef_property()
self.enforcer.enforce.assert_called_once_with(
self.context, 'modify_metadef_property', mock.ANY)
def test_remove_metadef_property(self):
self.policy.remove_metadef_property()
self.enforcer.enforce.assert_called_once_with(
self.context, 'remove_metadef_property', mock.ANY)
def test_add_metadef_resource_type_association(self):
self.policy.add_metadef_resource_type_association()
self.enforcer.enforce.assert_called_once_with(