openstack/glance
Code Issues Proposed changes

Merge "Add 'tasks_api_access' policy"

Browse Source
This commit is contained in:
Jenkins 2017-08-18 16:50:13 +00:00 committed by Gerrit Code Review
commit 7c0d34b73a
3 changed files with 98 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
etc/policy.json
View File

@ -26,10 +26,11 @@
"manage_image_cache": "role:admin",
"get_task": "role:admin",
"get_tasks": "role:admin",
"add_task": "role:admin",
"modify_task": "role:admin",
"get_task": "",
"get_tasks": "",
"add_task": "",
"modify_task": "",
"tasks_api_access": "role:admin",
"deactivate": "",
"reactivate": "",

20
glance/api/v2/tasks.py
View File

@ -67,6 +67,8 @@ class TasksController(object):
@debtcollector.removals.remove(message=_DEPRECATION_MESSAGE)
def create(self, req, task):
# NOTE(rosmaita): access to this call is enforced in the deserializer
task_factory = self.gateway.get_task_factory(req.context)
executor_factory = self.gateway.get_task_executor_factory(req.context)
task_repo = self.gateway.get_task_repo(req.context)
@ -88,6 +90,8 @@ class TasksController(object):
@debtcollector.removals.remove(message=_DEPRECATION_MESSAGE)
def index(self, req, marker=None, limit=None, sort_key='created_at',
sort_dir='desc', filters=None):
# NOTE(rosmaita): access to this call is enforced in the deserializer
result = {}
if filters is None:
filters = {}
@ -115,6 +119,7 @@ class TasksController(object):
@debtcollector.removals.remove(message=_DEPRECATION_MESSAGE)
def get(self, req, task_id):
_enforce_access_policy(self.policy, req)
try:
task_repo = self.gateway.get_task_repo(req.context)
task = task_repo.get(task_id)
@ -135,6 +140,7 @@ class TasksController(object):
@debtcollector.removals.remove(message=_DEPRECATION_MESSAGE)
def delete(self, req, task_id):
_enforce_access_policy(self.policy, req)
msg = (_("This operation is currently not permitted on Glance Tasks. "
"They are auto deleted after reaching the time based on "
"their expires_at property."))
@ -201,11 +207,14 @@ class RequestDeserializer(wsgi.JSONRequestDeserializer):
msg = _("Task '%s' is required") % param
raise webob.exc.HTTPBadRequest(explanation=msg)
def __init__(self, schema=None):
def __init__(self, schema=None, policy_engine=None):
super(RequestDeserializer, self).__init__()
self.schema = schema or get_task_schema()
# want to enforce the access policy as early as possible
self.policy_engine = policy_engine or policy.Enforcer()
def create(self, request):
_enforce_access_policy(self.policy_engine, request)
body = self._get_request_body(request)
self._validate_create_body(body)
try:
@ -222,6 +231,7 @@ class RequestDeserializer(wsgi.JSONRequestDeserializer):
return dict(task=task)
def index(self, request):
_enforce_access_policy(self.policy_engine, request)
params = request.params.copy()
limit = params.pop('limit', None)
marker = params.pop('marker', None)
@ -389,6 +399,14 @@ _TASK_SCHEMA = {
}
def _enforce_access_policy(policy_engine, request):
try:
policy_engine.enforce(request.context, 'tasks_api_access', {})
except exception.Forbidden:
LOG.debug("User does not have permission to access the Tasks API")
raise webob.exc.HTTPForbidden()
def get_task_schema():
properties = copy.deepcopy(_TASK_SCHEMA)
schema = glance.schema.Schema('task', properties)

74
glance/tests/unit/v2/test_tasks_resource.py
View File

@ -475,6 +475,14 @@ class TestTasksControllerPolicies(base.IsolatedUnitTest):
self.assertRaises(webob.exc.HTTPForbidden, self.controller.get,
request, task_id=UUID2)
def test_access_get_unauthorized(self):
rules = {"tasks_api_access": False,
"get_task": True}
self.policy.set_rules(rules)
request = unit_test_utils.get_fake_request()
self.assertRaises(webob.exc.HTTPForbidden, self.controller.get,
request, task_id=UUID2)
def test_create_task_unauthorized(self):
rules = {"add_task": False}
self.policy.set_rules(rules)
@ -490,6 +498,72 @@ class TestTasksControllerPolicies(base.IsolatedUnitTest):
request,
'fake_id')
def test_access_delete_unauthorized(self):
rules = {"tasks_api_access": False}
self.policy.set_rules(rules)
request = unit_test_utils.get_fake_request()
self.assertRaises(webob.exc.HTTPForbidden,
self.controller.delete,
request,
'fake_id')
class TestTasksDeserializerPolicies(test_utils.BaseTestCase):
# NOTE(rosmaita): this is a bit weird, but we check the access
# policy in the RequestDeserializer for calls that take bodies
# or query strings because we want to make sure the failure is
# a 403, not a 400 due to bad request format
def setUp(self):
super(TestTasksDeserializerPolicies, self).setUp()
self.policy = unit_test_utils.FakePolicyEnforcer()
self.deserializer = glance.api.v2.tasks.RequestDeserializer(
schema=None, policy_engine=self.policy)
bad_path = '/tasks?limit=NaN'
def test_access_index_authorized_bad_query_string(self):
"""Allow access, fail with 400"""
rules = {"tasks_api_access": True,
"get_tasks": True}
self.policy.set_rules(rules)
request = unit_test_utils.get_fake_request(self.bad_path)
self.assertRaises(webob.exc.HTTPBadRequest, self.deserializer.index,
request)
def test_access_index_unauthorized(self):
"""Disallow access with bad request, fail with 403"""
rules = {"tasks_api_access": False,
"get_tasks": True}
self.policy.set_rules(rules)
request = unit_test_utils.get_fake_request(self.bad_path)
self.assertRaises(webob.exc.HTTPForbidden, self.deserializer.index,
request)
bad_task = {'typo': 'import', 'input': {"import_from": "fake"}}
def test_access_create_authorized_bad_format(self):
"""Allow access, fail with 400"""
rules = {"tasks_api_access": True,
"add_task": True}
self.policy.set_rules(rules)
request = unit_test_utils.get_fake_request()
request.body = jsonutils.dump_as_bytes(self.bad_task)
self.assertRaises(webob.exc.HTTPBadRequest,
self.deserializer.create,
request)
def test_access_create_unauthorized(self):
"""Disallow access with bad request, fail with 403"""
rules = {"tasks_api_access": False,
"add_task": True}
self.policy.set_rules(rules)
request = unit_test_utils.get_fake_request()
request.body = jsonutils.dump_as_bytes(self.bad_task)
self.assertRaises(webob.exc.HTTPForbidden,
self.deserializer.create,
request)
class TestTasksDeserializer(test_utils.BaseTestCase):