"object_assignments": {
"object_security_level": {
- "servers": ["low"],
- "vm1": ["low"],
- "vm2": ["medium"],
- "file1": ["low"],
- "file2": ["medium"]
+ "servers": ["low"]
},
"type": {
- "servers": ["computing"],
- "vm1": ["computing"],
- "vm2": ["computing"],
- "file1": ["storage"],
- "file2": ["storage"]
+ "servers": ["computing"]
},
"object_id": {
- "servers": ["servers"],
- "vm1": ["vm1"],
- "vm2": ["vm2"],
- "file1": ["file1"],
- "file2": ["file2"]
+ "servers": ["servers"]
}
}
}
],
"rbac_rule":[
["dev", "xx", "read", "servers"],
- ["dev", "xx", "read", "vm1"],
- ["dev", "xx", "read", "vm2"],
- ["dev", "xx", "read", "file1"],
- ["dev", "xx", "read", "file2"],
- ["dev", "xx", "write", "vm1"],
- ["dev", "xx", "write", "vm2"],
- ["dev", "xx", "write", "file1"],
- ["dev", "xx", "write", "file2"],
["admin", "xx", "read", "servers"],
- ["admin", "ft", "read", "servers"],
- ["admin", "ft", "read", "vm1"],
- ["admin", "ft", "read", "vm2"],
- ["admin", "ft", "read", "file1"],
- ["admin", "ft", "read", "file2"],
- ["admin", "ft", "write", "vm1"],
- ["admin", "ft", "write", "vm2"],
- ["admin", "ft", "write", "file1"],
- ["admin", "ft", "write", "file2"]
+ ["admin", "ft", "read", "servers"]
]
}
import itertools
+from oslo_log import log
+LOG = log.getLogger(__name__)
+
""" an example of authz_buffer, sub_meta_rule_dict, rule_dict
authz_buffer = {
super(Authz_v3, self).__init__()
@controller.protected()
- def get_authz(self, context, tenant_name, subject_name, object_name, action_name):
+ def get_authz(self, context, tenant_id, subject_k_id, object_name, action_name):
try:
- return self.authz_api.authz(tenant_name, subject_name, object_name, action_name)
- except:
- return False
+ return self.authz_api.authz(tenant_id, subject_k_id, object_name, action_name)
+ except Exception as e:
+ return {'authz': False, 'comment': unicode(e)}
@dependency.requires('admin_api', 'authz_api')
decision = one_true(decision_buffer)
if not decision:
raise AuthzException("{} {}-{}-{}".format(intra_extension_id, subject_id, action_id, object_id))
- return decision
+ return {'authz': decision, 'comment': ''}
@enforce("read", "intra_extensions")
def get_intra_extensions_dict(self, user_id):
def __init__(self):
super(IntraExtensionAuthzManager, self).__init__()
- def authz(self, tenant_name, subject_name, object_name, action_name, genre="authz"):
+ def authz(self, tenant_id, subject_k_id, object_name, action_name, genre="authz"):
"""Check authorization for a particular action.
:return: True or False or raise an exception
"""
genre = "intra_admin_extension_id"
tenants_dict = self.tenant_api.get_tenants_dict(self.root_api.get_root_admin_id())
- tenant_id = None
- for _tenant_id in tenants_dict:
- if tenants_dict[_tenant_id]["name"] == tenant_name:
- tenant_id = _tenant_id
- break
- if not tenant_id:
- raise TenantUnknown
+
+ if tenant_id not in tenants_dict:
+ raise TenantUnknown()
intra_extension_id = tenants_dict[tenant_id][genre]
if not intra_extension_id:
raise TenantNoIntraExtension()
-
subjects_dict = self.driver.get_subjects_dict(intra_extension_id)
subject_id = None
for _subject_id in subjects_dict:
- if subjects_dict[_subject_id]['keystone_name'] == subject_name:
- # subject_id = subjects_dict[_subject_id]['keystone_id']
+ if subjects_dict[_subject_id]['keystone_id'] == subject_k_id:
subject_id = _subject_id
break
if not subject_id:
raise SubjectUnknown()
-
objects_dict = self.driver.get_objects_dict(intra_extension_id)
object_id = None
for _object_id in objects_dict:
# Authz route
self._add_resource(
mapper, authz_controller,
- path=self.PATH_PREFIX+'/authz/{tenant_name}/{subject_name}/{object_name}/{action_name}',
+ path=self.PATH_PREFIX+'/authz/{tenant_id}/{subject_k_id}/{object_name}/{action_name}',
get_action='get_authz',
rel=self._get_rel('authz'),
path_vars={
- 'tenant_name': self._get_path('tenants'),
- 'subject_name': self._get_path('subjects'),
+ 'tenant_id': self._get_path('tenants'),
+ 'subject_k_id': self._get_path('subjects'),
'object_name': self._get_path('objects'),
'action_name': self._get_path('actions'),
})
objects_dict = self.authz_manager.get_objects_dict(admin_subject_id, authz_ie_dict["id"])
- object_vm1_id = None
- object_vm2_id = None
- for _object_id in objects_dict:
- if objects_dict[_object_id]['name'] == 'vm1':
- object_vm1_id = _object_id
- if objects_dict[_object_id]['name'] == 'vm2':
- object_vm2_id = _object_id
+ object_vm1 = self.admin_manager.add_object_dict(admin_subject_id, authz_ie_dict["id"], {"name": "vm1", "description": "vm1"})
+ object_vm2 = self.admin_manager.add_object_dict(admin_subject_id, authz_ie_dict["id"], {"name": "vm2", "description": "vm2"})
+ object_vm1_id = object_vm1.keys()[0]
+ object_vm2_id = object_vm2.keys()[0]
if not object_vm1_id or not object_vm2_id:
raise Exception("Cannot run tests, database is corrupted ? (need upload and list in objects)")
objects_dict = self.authz_manager.get_objects_dict(admin_subject_id, authz_ie_dict["id"])
- object_vm1_id = None
- object_vm2_id = None
- for _object_id in objects_dict:
- if objects_dict[_object_id]['name'] == 'vm1':
- object_vm1_id = _object_id
- if objects_dict[_object_id]['name'] == 'vm2':
- object_vm2_id = _object_id
+ object_vm1 = self.admin_manager.add_object_dict(admin_subject_id, authz_ie_dict["id"], {"name": "vm1", "description": "vm1"})
+ object_vm2 = self.admin_manager.add_object_dict(admin_subject_id, authz_ie_dict["id"], {"name": "vm2", "description": "vm2"})
+ object_vm1_id = object_vm1.keys()[0]
+ object_vm2_id = object_vm2.keys()[0]
if not object_vm1_id or not object_vm2_id:
raise Exception("Cannot run tests, database is corrupted ? (need upload and list in objects)")
objects_dict = self.authz_manager.get_objects_dict(admin_subject_id, authz_ie_dict["id"])
- object_vm1_id = None
- object_vm2_id = None
- for _object_id in objects_dict:
- if objects_dict[_object_id]['name'] == 'vm1':
- object_vm1_id = _object_id
- if objects_dict[_object_id]['name'] == 'vm2':
- object_vm2_id = _object_id
+ object_vm1 = self.admin_manager.add_object_dict(admin_subject_id, authz_ie_dict["id"], {"name": "vm1", "description": "vm1"})
+ object_vm2 = self.admin_manager.add_object_dict(admin_subject_id, authz_ie_dict["id"], {"name": "vm2", "description": "vm2"})
+ object_vm1_id = object_vm1.keys()[0]
+ object_vm2_id = object_vm2.keys()[0]
if not object_vm1_id or not object_vm2_id:
raise Exception("Cannot run tests, database is corrupted ? (need upload and list in objects)")
self.assertRaises(
SubjectUnknown,
self.authz_manager.authz,
- tenant["name"], uuid.uuid4().hex, uuid.uuid4().hex, uuid.uuid4().hex
+ tenant["id"], uuid.uuid4().hex, uuid.uuid4().hex, uuid.uuid4().hex
)
# Test when subject is known but not the object
self.assertRaises(
ObjectUnknown,
self.authz_manager.authz,
- tenant["name"], demo_subject_dict["name"], uuid.uuid4().hex, uuid.uuid4().hex
+ tenant["id"], demo_subject_dict["keystone_id"], uuid.uuid4().hex, uuid.uuid4().hex
)
# Test when subject and object are known but not the action
self.assertRaises(
ActionUnknown,
self.authz_manager.authz,
- tenant["name"], demo_subject_dict["name"], my_object["name"], uuid.uuid4().hex
+ tenant["id"], demo_subject_dict["keystone_id"], my_object["name"], uuid.uuid4().hex
)
# Test when subject and object and action are known
self.assertRaises(
AuthzException,
self.authz_manager.authz,
- tenant["name"], demo_subject_dict["name"], my_object["name"], my_action["name"]
+ tenant["id"], demo_subject_dict["keystone_id"], my_object["name"], my_action["name"]
)
# Add a subject scope and test ObjectCategoryAssignmentOutOfScope
self.assertRaises(
AuthzException,
self.authz_manager.authz,
- tenant["name"], demo_subject_dict["name"], my_object["name"], my_action["name"]
+ tenant["id"], demo_subject_dict["keystone_id"], my_object["name"], my_action["name"]
)
# Add an object scope and test ActionCategoryAssignmentOutOfScope
self.assertRaises(
AuthzException,
self.authz_manager.authz,
- tenant["name"], demo_subject_dict["name"], my_object["name"], my_action["name"]
+ tenant["id"], demo_subject_dict["keystone_id"], my_object["name"], my_action["name"]
)
# Add an action scope and test SubjectCategoryAssignmentUnknown
self.assertRaises(
AuthzException,
self.authz_manager.authz,
- tenant["name"], demo_subject_dict["name"], my_object["name"], my_action["name"]
+ tenant["id"], demo_subject_dict["keystone_id"], my_object["name"], my_action["name"]
)
# Add a subject assignment and test ObjectCategoryAssignmentUnknown
self.assertRaises(
AuthzException,
self.authz_manager.authz,
- tenant["name"], demo_subject_dict["name"], my_object["name"], my_action["name"]
+ tenant["id"], demo_subject_dict["keystone_id"], my_object["name"], my_action["name"]
)
# Add an object assignment and test ActionCategoryAssignmentUnknown
self.assertRaises(
AuthzException,
self.authz_manager.authz,
- tenant["name"], demo_subject_dict["name"], my_object["name"], my_action["name"]
+ tenant["id"], demo_subject_dict["keystone_id"], my_object["name"], my_action["name"]
)
# Add an action assignment and test RuleUnknown
self.assertRaises(
AuthzException,
self.authz_manager.authz,
- tenant["name"], admin_subject_dict["name"], my_object["name"], my_action["name"]
+ tenant["id"], admin_subject_dict["keystone_id"], my_object["name"], my_action["name"]
)
# Add the correct rule and test that no exception is raised
authz_ie_dict["id"]
)
- print("authz_ie_dict[\"id\"]", authz_ie_dict["id"])
self.assertRaises(
SubMetaRuleAlgorithmNotExisting,
self.admin_manager.add_sub_meta_rule_dict,
self.assertRaises(
AuthzException,
self.authz_manager.authz,
- tenant["name"], admin_subject_dict["name"], my_object["name"], my_action["name"]
+ tenant["id"], admin_subject_dict["keystone_id"], my_object["name"], my_action["name"]
)
- result = self.authz_manager.authz(tenant["name"], demo_subject_dict["name"], my_object["name"], my_action["name"])
- self.assertEqual(True, result)
+ result = self.authz_manager.authz(tenant["id"], demo_subject_dict["keystone_id"], my_object["name"], my_action["name"])
+ self.assertIsInstance(result, dict)
+ self.assertIn('authz', result)
+ self.assertEquals(result['authz'], True)
def test_subjects(self):
authz_ie_dict = create_intra_extension(self, "policy_authz")
objects_dict = self.authz_manager.get_objects_dict(admin_subject_id, authz_ie_dict["id"])
- object_vm1_id = None
- object_vm2_id = None
- for _object_id in objects_dict:
- if objects_dict[_object_id]['name'] == 'vm1':
- object_vm1_id = _object_id
- if objects_dict[_object_id]['name'] == 'vm2':
- object_vm2_id = _object_id
+ object_vm1 = self.admin_manager.add_object_dict(admin_subject_id, authz_ie_dict["id"], {"name": "vm1", "description": "vm1"})
+ object_vm2 = self.admin_manager.add_object_dict(admin_subject_id, authz_ie_dict["id"], {"name": "vm2", "description": "vm2"})
+ object_vm1_id = object_vm1.keys()[0]
+ object_vm2_id = object_vm2.keys()[0]
if not object_vm1_id or not object_vm2_id:
raise Exception("Cannot run tests, database is corrupted ? (need upload and list in objects)")
:return: the action or ""
"""
action = ""
+ self.input = ""
if component == "nova":
length = int(env.get('CONTENT_LENGTH', '0'))
# TODO (dthom): compute for Nova, Cinder, Neutron, ...
if length > 0:
try:
sub_action_object = env['wsgi.input'].read(length)
+ self.input = sub_action_object
action = json.loads(sub_action_object).keys()[0]
body = StringIO(sub_action_object)
env['wsgi.input'] = body
@staticmethod
def _get_object(env, component):
if component == "nova":
- # get the object ID which is located before "action" in the URL
- return env.get("PATH_INFO").split("/")[-2]
+ # http://developer.openstack.org/api-ref-compute-v2.1.html
+ # nova URLs:
+ # /<tenant_id>/servers/<server_id>
+ # list details for server_id
+ # /<tenant_id>/servers/<server_id>/action
+ # execute action to server_id
+ # /<tenant_id>/servers/<server_id>/metadata
+ # show metadata from server_id
+ # /<tenant_id>/servers/details
+ # list servers
+ url = env.get("PATH_INFO").split("/")
+ if url[-1] == "detail":
+ return "servers"
+ try:
+ return url[3]
+ except IndexError:
+ return
elif component == "swift":
# remove the "/v1/" part of the URL
return env.get("PATH_INFO").split("/", 2)[-1].replace("/", "-")
component = self._find_openstack_component(env)
action_id = self._get_action(env, component)
if action_id:
- self._LOG.debug("OpenStack component {}".format(component))
object_id = self._get_object(env, component)
- self._LOG.debug("{}-{}-{}-{}".format(subject_id, object_id, action_id, tenant_id))
+ if not object_id:
+ object_id = "servers"
self.__set_token()
resp = self._get_authz_from_moon(self.x_subject_token, tenant_id, subject_id, object_id, action_id)
- self._LOG.info("Moon answer: {}-{}".format(resp.status_code, resp.content))
self.__unset_token()
if resp.status_code == 200:
try:
self._LOG.debug(answer)
if "authz" in answer and answer["authz"]:
return self._app(env, start_response)
- except:
- raise exception.Unauthorized(message="You are not authorized to do that!")
+ except Exception as e:
+ # self._LOG.error("You are not authorized to do that!")
+ raise exception.Unauthorized(message="You are not authorized to do that! ({})".format(unicode(e)))
self._LOG.debug("No action_id found for {}".format(env.get("PATH_INFO")))
# If action is not found, we can't raise an exception because a lots of action is missing
# in function self._get_action, it is not possible to get them all.
Code Review / moon.git / commitdiff