x/python-ganttclient
Code Issues Proposed changes

Add policy checks to Compute.API

Browse Source 
* Second step of blueprint interim-nova-authz-service
 * Adds policy.json to define policy
 * Add nova.policy.wrap_enforce decorator
 * wrap majority of compute api functions with wrap_enforce

Change-Id: If6702873db3249921f931a42e889ee7d0338e4b8
This commit is contained in:
Brian Waldon
2011-12-30 13:11:56 -08:00
parent afbf369af3
commit 117ead3576
3 changed files with 190 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
nova/common/policy.py
View File

@@ -22,7 +22,7 @@ import urllib
import urllib2
class NotAllowed(Exception):
class NotAuthorized(Exception):
pass
@@ -91,14 +91,14 @@ def enforce(match_list, target_dict, credentials_dict):
Credentials dicts contain as much information as we can about the user
performing the action.
:raises NotAllowed if the check fails
:raises NotAuthorized if the check fails
"""
global _BRAIN
if not _BRAIN:
_BRAIN = Brain()
if not _BRAIN.check(match_list, target_dict, credentials_dict):
raise NotAllowed()
raise NotAuthorized()
class Brain(object):

77
nova/tests/policy.json
View File

@@ -1,11 +1,70 @@
{
"true" : [],
"compute:create_instance" : [],
"compute:attach_network" : [],
"compute:attach_volume" : [],
"compute:list_instances": [],
"compute:get_instance": [],
"network:attach_network" : [],
"volume:create_volume": [],
"volume:attach_volume": []
"compute:create": [],
"compute:create:attach_network": [],
"compute:create:attach_volume": [],
"compute:get": [],
"compute:get_all" :[],
"compute:update": [],
"compute:get_instance_metadata": [],
"compute:update_instance_metadata": [],
"compute:delete_instance_metadata": [],
"compute:get_instance_faults": [],
"compute:get_actions": [],
"compute:get_diagnostics": [],
"compute:get_lock": [],
"compute:lock": [],
"compute:unlock": [],
"compute:get_ajax_console": [],
"compute:get_vnc_console": [],
"compute:get_console_output": [],
"compute:associate_floating_ip": [],
"compute:reset_network": [],
"compute:inject_network_info": [],
"compute:add_fixed_ip": [],
"compute:remove_fixed_ip": [],
"compute:attach_volume": [],
"compute:detach_volume": [],
"compute:inject_file": [],
"compute:set_admin_password": [],
"compute:rescue": [],
"compute:unrescue": [],
"compute:suspend": [],
"compute:resume": [],
"compute:pause": [],
"compute:unpause": [],
"compute:start": [],
"compute:stop": [],
"compute:resize": [],
"compute:confirm_resize": [],
"compute:revert_resize": [],
"compute:rebuild": [],
"compute:reboot": [],
"compute:snapshot": [],
"compute:backup": [],
"compute:add_security_group": [],
"compute:remove_security_group": [],
"compute:delete": [],
"compute:soft_delete": [],
"compute:force_delete": [],
"compute:restore": []
}

123
nova/tests/test_compute.py
View File

@@ -28,7 +28,9 @@ import mox
import webob.exc
import nova
import nova.common.policy
from nova import compute
import nova.compute.api
from nova.compute import instance_types
from nova.compute import manager as compute_manager
from nova.compute import power_state
@@ -42,8 +44,9 @@ from nova.image import fake as fake_image
from nova import log as logging
from nova.network.quantum import client as quantum_client
from nova.notifier import test_notifier
from nova.scheduler import driver as scheduler_driver
import nova.policy
from nova import rpc
from nova.scheduler import driver as scheduler_driver
from nova import test
from nova.tests import fake_network
from nova import utils
@@ -111,7 +114,8 @@ class BaseTestCase(test.TestCase):
self.compute = utils.import_object(FLAGS.compute_manager)
self.user_id = 'fake'
self.project_id = 'fake'
self.context = context.RequestContext(self.user_id, self.project_id)
self.context = context.RequestContext(self.user_id,
self.project_id)
test_notifier.NOTIFICATIONS = []
self.mox = mox.Mox()
self.total_waits = 0
@@ -878,7 +882,9 @@ class ComputeTestCase(BaseTestCase):
instance_uuid = instance['uuid']
self.compute.run_instance(self.context, instance_uuid)
non_admin_context = context.RequestContext(None, None, is_admin=False)
non_admin_context = context.RequestContext(None,
None,
is_admin=False)
# decorator should return False (fail) with locked nonadmin context
self.compute.lock_instance(self.context, instance_uuid)
@@ -2815,7 +2821,7 @@ class ComputeAPITestCase(BaseTestCase):
def test_attach_volume_invalid(self):
self.assertRaises(exception.ApiError,
self.compute_api.attach_volume,
None,
self.context,
None,
None,
'/dev/invalid')
@@ -2966,3 +2972,112 @@ class ComputeAPITestCase(BaseTestCase):
self.compute_api.inject_file(self.context, instance,
"/tmp/test", "File Contents")
db.instance_destroy(self.context, instance['id'])
class ComputePolicyTestCase(BaseTestCase):
def setUp(self):
super(ComputePolicyTestCase, self).setUp()
nova.policy.reset()
nova.policy.init()
self.compute_api = compute.API()
def tearDown(self):
super(ComputePolicyTestCase, self).tearDown()
nova.policy.reset()
def _set_rules(self, rules):
nova.common.policy.set_brain(nova.common.policy.HttpBrain(rules))
def test_actions_are_prefixed(self):
self.mox.StubOutWithMock(nova.policy, 'enforce')
nova.policy.enforce(self.context, 'compute:reboot', {})
self.mox.ReplayAll()
nova.compute.api.check_policy(self.context, 'reboot', {})
self.mox.UnsetStubs()
self.mox.VerifyAll()
def test_wrapped_method(self):
instance = self._create_fake_instance()
self.compute.run_instance(self.context, instance['uuid'])
# force delete to fail
rules = {"compute:delete": [["false:false"]]}
self._set_rules(rules)
self.assertRaises(exception.PolicyNotAuthorized,
self.compute_api.delete, self.context, instance)
# reset rules to allow deletion
rules = {"compute:delete": []}
self._set_rules(rules)
self.compute_api.delete(self.context, instance)
def test_create_fail(self):
rules = {"compute:create": [["false:false"]]}
self._set_rules(rules)
self.assertRaises(exception.PolicyNotAuthorized,
self.compute_api.create, self.context, '1', '1')
def test_create_attach_volume_fail(self):
rules = {
"compute:create": [],
"compute:create:attach_network": [["false:false"]],
"compute:create:attach_volume": [],
}
self._set_rules(rules)
self.assertRaises(exception.PolicyNotAuthorized,
self.compute_api.create, self.context, '1', '1',
requested_networks='blah',
block_device_mapping='blah')
def test_create_attach_network_fail(self):
rules = {
"compute:create": [],
"compute:create:attach_network": [],
"compute:create:attach_volume": [["false:false"]],
}
self._set_rules(rules)
self.assertRaises(exception.PolicyNotAuthorized,
self.compute_api.create, self.context, '1', '1',
requested_networks='blah',
block_device_mapping='blah')
def test_get_fail(self):
instance = self._create_fake_instance()
rules = {
"compute:get": [["false:false"]],
}
self._set_rules(rules)
self.assertRaises(exception.PolicyNotAuthorized,
self.compute_api.get, self.context, instance['uuid'])
def test_get_all_fail(self):
rules = {
"compute:get_all": [["false:false"]],
}
self._set_rules(rules)
self.assertRaises(exception.PolicyNotAuthorized,
self.compute_api.get_all, self.context)
def test_get_instance_faults(self):
instance1 = self._create_fake_instance()
instance2 = self._create_fake_instance()
instances = [instance1, instance2]
rules = {
"compute:get_instance_faults": [["false:false"]],
}
self._set_rules(rules)
self.assertRaises(exception.PolicyNotAuthorized,
self.compute_api.get_instance_faults,
self.context, instances)