summaryrefslogtreecommitdiffstats
path: root/keystone-moon/keystone
diff options
context:
space:
mode:
authorasteroide <thomas.duval@orange.com>2015-07-10 17:35:34 +0200
committerasteroide <thomas.duval@orange.com>2015-07-13 22:05:16 +0200
commit7278636073202990ad1775819ae144dfb766367a (patch)
treee0f585db68731f5e14d4354c2f284777c9998e9c /keystone-moon/keystone
parent4f99262f0f2c6f563ac2a376d09646220508c625 (diff)
Hamonize the uses of UUID and name in core.py.
Change-Id: I15e3c2e8a3f3ce5778bb8366c78eb2657b317686
Diffstat (limited to 'keystone-moon/keystone')
-rw-r--r--keystone-moon/keystone/contrib/moon/backends/sql.py39
-rw-r--r--keystone-moon/keystone/contrib/moon/core.py664
-rw-r--r--keystone-moon/keystone/contrib/moon/exception.py42
-rw-r--r--keystone-moon/keystone/tests/moon/unit/test_unit_core_intra_extension_admin.py54
-rw-r--r--keystone-moon/keystone/tests/moon/unit/test_unit_core_intra_extension_authz.py78
5 files changed, 513 insertions, 364 deletions
diff --git a/keystone-moon/keystone/contrib/moon/backends/sql.py b/keystone-moon/keystone/contrib/moon/backends/sql.py
index 361e53ed..35884a93 100644
--- a/keystone-moon/keystone/contrib/moon/backends/sql.py
+++ b/keystone-moon/keystone/contrib/moon/backends/sql.py
@@ -424,7 +424,7 @@ class IntraExtensionConnector(IntraExtensionDriver):
for attr in Subject.attributes:
if attr != 'id':
setattr(ref, attr, getattr(new_ref, attr))
- return {"subject": {"uuid": subject_uuid, "name": subject_name}}
+ return ref.to_dict()
def remove_subject(self, extension_uuid, subject_uuid):
with sql.transaction() as session:
@@ -502,7 +502,7 @@ class IntraExtensionConnector(IntraExtensionDriver):
for attr in Object.attributes:
if attr != 'id':
setattr(ref, attr, getattr(new_ref, attr))
- return {"object": {"uuid": object_uuid, "name": object_name}}
+ return ref.to_dict()
def remove_object(self, extension_uuid, object_uuid):
with sql.transaction() as session:
@@ -580,7 +580,7 @@ class IntraExtensionConnector(IntraExtensionDriver):
for attr in Action.attributes:
if attr != 'id':
setattr(ref, attr, getattr(new_ref, attr))
- return {"action": {"uuid": action_uuid, "name": action_name}}
+ return ref.to_dict()
def remove_action(self, extension_uuid, action_uuid):
with sql.transaction() as session:
@@ -660,7 +660,7 @@ class IntraExtensionConnector(IntraExtensionDriver):
for attr in SubjectCategory.attributes:
if attr != 'id':
setattr(ref, attr, getattr(new_ref, attr))
- return {"subject_category": {"uuid": subject_category_uuid, "name": subject_category_name}}
+ return ref.to_dict()
def remove_subject_category(self, extension_uuid, subject_category_uuid):
with sql.transaction() as session:
@@ -741,7 +741,7 @@ class IntraExtensionConnector(IntraExtensionDriver):
for attr in ObjectCategory.attributes:
if attr != 'id':
setattr(ref, attr, getattr(new_ref, attr))
- return {"object_category": {"uuid": object_category_uuid, "name": object_category_name}}
+ return ref.to_dict()
def remove_object_category(self, extension_uuid, object_category_uuid):
with sql.transaction() as session:
@@ -822,7 +822,7 @@ class IntraExtensionConnector(IntraExtensionDriver):
for attr in ActionCategory.attributes:
if attr != 'id':
setattr(ref, attr, getattr(new_ref, attr))
- return {"action_category": {"uuid": action_category_uuid, "name": action_category_name}}
+ return ref.to_dict()
def remove_action_category(self, extension_uuid, action_category_uuid):
with sql.transaction() as session:
@@ -862,8 +862,7 @@ class IntraExtensionConnector(IntraExtensionDriver):
raise IntraExtensionNotFound()
result = copy.deepcopy(ref.to_dict())
if subject_category not in result["subject_category_scope"].keys():
- raise AuthzMetadata()
- result["subject_category_scope"] = {subject_category: result["subject_category_scope"][subject_category]}
+ raise SubjectScopeUnknown()
return result
def set_subject_category_scope_dict(self, extension_uuid, subject_category, scope):
@@ -880,14 +879,13 @@ class IntraExtensionConnector(IntraExtensionDriver):
}
)
session.add(new_ref)
- ref = new_ref
else:
tmp_ref = ref.to_dict()
tmp_ref['subject_category_scope'].update({subject_category: scope})
session.delete(ref)
new_ref = SubjectCategoryScope.from_dict(tmp_ref)
session.add(new_ref)
- return ref.to_dict()
+ return new_ref.to_dict()
def add_subject_category_scope_dict(self, extension_uuid, subject_category, scope_uuid, scope_name):
with sql.transaction() as session:
@@ -901,8 +899,7 @@ class IntraExtensionConnector(IntraExtensionDriver):
if subject_category not in scope.keys():
scope[subject_category] = dict()
scope[subject_category][scope_uuid] = scope_name
- self.set_subject_category_scope_dict(extension_uuid, subject_category, scope[subject_category])
- return {"subject_category_scope": {"uuid": scope_uuid, "name": scope_name}}
+ return self.set_subject_category_scope_dict(extension_uuid, subject_category, scope[subject_category])
def remove_subject_category_scope_dict(self, extension_uuid, subject_category, scope_uuid):
with sql.transaction() as session:
@@ -942,8 +939,7 @@ class IntraExtensionConnector(IntraExtensionDriver):
raise IntraExtensionNotFound()
result = copy.deepcopy(ref.to_dict())
if object_category not in result["object_category_scope"].keys():
- raise AuthzMetadata()
- result["object_category_scope"] = {object_category: result["object_category_scope"][object_category]}
+ raise ObjectScopeUnknown()
return result
def set_object_category_scope_dict(self, extension_uuid, object_category, scope):
@@ -960,14 +956,13 @@ class IntraExtensionConnector(IntraExtensionDriver):
}
)
session.add(new_ref)
- ref = new_ref
else:
tmp_ref = ref.to_dict()
tmp_ref['object_category_scope'].update({object_category: scope})
session.delete(ref)
new_ref = ObjectCategoryScope.from_dict(tmp_ref)
session.add(new_ref)
- return ref.to_dict()
+ return new_ref.to_dict()
def add_object_category_scope_dict(self, extension_uuid, object_category, scope_uuid, scope_name):
with sql.transaction() as session:
@@ -981,8 +976,7 @@ class IntraExtensionConnector(IntraExtensionDriver):
if object_category not in scope:
scope[object_category] = dict()
scope[object_category][scope_uuid] = scope_name
- self.set_object_category_scope_dict(extension_uuid, object_category, scope[object_category])
- return {"object_category_scope": {"uuid": scope_uuid, "name": scope_name}}
+ return self.set_object_category_scope_dict(extension_uuid, object_category, scope[object_category])
def remove_object_category_scope_dict(self, extension_uuid, object_category, scope_uuid):
with sql.transaction() as session:
@@ -1022,8 +1016,7 @@ class IntraExtensionConnector(IntraExtensionDriver):
raise IntraExtensionNotFound()
result = copy.deepcopy(ref.to_dict())
if action_category not in result["action_category_scope"].keys():
- raise AuthzMetadata("Unknown category id {}/{}".format(action_category, result["action_category_scope"].keys()))
- result["action_category_scope"] = {action_category: result["action_category_scope"][action_category]}
+ raise ActionScopeUnknown()
return result
def set_action_category_scope_dict(self, extension_uuid, action_category, scope):
@@ -1040,14 +1033,13 @@ class IntraExtensionConnector(IntraExtensionDriver):
}
)
session.add(new_ref)
- ref = new_ref
else:
tmp_ref = ref.to_dict()
tmp_ref['action_category_scope'].update({action_category: scope})
session.delete(ref)
new_ref = ActionCategoryScope.from_dict(tmp_ref)
session.add(new_ref)
- return ref.to_dict()
+ return new_ref.to_dict()
def add_action_category_scope_dict(self, extension_uuid, action_category, scope_uuid, scope_name):
with sql.transaction() as session:
@@ -1061,8 +1053,7 @@ class IntraExtensionConnector(IntraExtensionDriver):
if action_category not in scope:
scope[action_category] = dict()
scope[action_category][scope_uuid] = scope_name
- self.set_action_category_scope_dict(extension_uuid, action_category, scope[action_category])
- return {"action_category_scope": {"uuid": scope_uuid, "name": scope_name}}
+ return self.set_action_category_scope_dict(extension_uuid, action_category, scope[action_category])
def remove_action_category_scope_dict(self, extension_uuid, action_category, scope_uuid):
with sql.transaction() as session:
diff --git a/keystone-moon/keystone/contrib/moon/core.py b/keystone-moon/keystone/contrib/moon/core.py
index 3bf3a13d..ad6bf93d 100644
--- a/keystone-moon/keystone/contrib/moon/core.py
+++ b/keystone-moon/keystone/contrib/moon/core.py
@@ -90,7 +90,7 @@ def enforce(actions, object, **extra):
actions = _actions
for action in actions:
if self.admin_api.authz(
- intra_extension_uuid,
+ _admin_extension_uuid,
user_name,
object,
action):
@@ -243,7 +243,7 @@ class TenantManager(manager.Manager):
def get_admin_extension_uuid(self, authz_extension_uuid):
_tenants = self.get_tenant_dict()
for _tenant_uuid in _tenants:
- if authz_extension_uuid == _tenants[_tenant_uuid]['authz']and _tenants[_tenant_uuid]['admin']:
+ if authz_extension_uuid == _tenants[_tenant_uuid]['authz'] and _tenants[_tenant_uuid]['admin']:
return _tenants[_tenant_uuid]['admin']
self.moonlog_api.error(_("No IntraExtension found mapping this Authz IntraExtension: {}.".format(
authz_extension_uuid)))
@@ -318,13 +318,25 @@ class IntraExtensionManager(manager.Manager):
_authz_buffer['object_uuid'] = object_uuid
_authz_buffer['action_uuid'] = action_uuid
- try:
- _meta_data_dict = self.driver.get_meta_data_dict(intra_extension_uuid)
- _subject_assignment_dict = self.driver.get_subject_category_assignment_dict(intra_extension_uuid, subject_uuid)
- _object_assignment_dict = self.driver.get_object_category_assignment_dict(intra_extension_uuid, object_uuid)
- _action_assignment_dict = self.driver.get_action_category_assignment_dict(intra_extension_uuid, action_uuid)
- except exception: # Execption for ItemUnknow, ItemCategoryAssignmentOutOfScope, ItemCategoryAssignmentUnknown
- pass
+ _meta_data_dict = {}
+ _meta_data_dict["subject_categories"] = self.driver.get_subject_category_dict(intra_extension_uuid)["subject_categories"]
+ _meta_data_dict["object_categories"] = self.driver.get_object_category_dict(intra_extension_uuid)["object_categories"]
+ _meta_data_dict["action_categories"] = self.driver.get_action_category_dict(intra_extension_uuid)["action_categories"]
+
+ _subject_assignment_dict = dict()
+ for category in _meta_data_dict["subject_categories"]:
+ _subject_assignment_dict[category] = self.driver.get_subject_category_assignment_dict(
+ intra_extension_uuid, category)["subject_category_assignments"]
+
+ _object_assignment_dict = dict()
+ for category in _meta_data_dict["object_categories"]:
+ _object_assignment_dict[category] = self.driver.get_object_category_assignment_dict(
+ intra_extension_uuid, category)["object_category_assignments"]
+
+ _action_assignment_dict = dict()
+ for category in _meta_data_dict["action_categories"]:
+ _action_assignment_dict[category] = self.driver.get_action_category_assignment_dict(
+ intra_extension_uuid, category)["action_category_assignments"]
_authz_buffer['subject_attributes'] = dict()
_authz_buffer['object_attributes'] = dict()
@@ -360,26 +372,26 @@ class IntraExtensionManager(manager.Manager):
_authz_buffer = self.__get_authz_buffer(intra_extension_uuid, subject_uuid, object_uuid, action_uuid)
_decision_buffer = dict()
- try:
- _meta_rule_dict = self.driver.get_meta_rule_dict(intra_extension_uuid)
- _rule_dict = self.driver.get_rule_dict(intra_extension_uuid)
- except exception: # Execption for rule
- pass
+ _meta_rule_dict = self.driver.get_meta_rule_dict(intra_extension_uuid)
+ _rule_dict = self.driver.get_rule_dict(intra_extension_uuid)
for _rule in _meta_rule_dict['sub_meta_rules']:
if _meta_rule_dict['sub_meta_rules'][_rule]['algorithm'] == 'inclusion':
- _decision_buffer[_rule] = algo_inclusion(_authz_buffer, _meta_rule_dict['sub_meta_rules'][_rule], _rule_dict[_rule])
+ _decision_buffer[_rule] = algo_inclusion(
+ _authz_buffer,
+ _meta_rule_dict['sub_meta_rules'][_rule],
+ _rule_dict['rules'][_rule])
elif _meta_rule_dict['sub_meta_rules'][_rule]['algorithm'] == 'comparison':
- _decision_buffer[_rule] = algo_comparison(_authz_buffer, _meta_rule_dict['sub_meta_rules'][_rule], _rule_dict[_rule])
+ _decision_buffer[_rule] = algo_comparison(
+ _authz_buffer,
+ _meta_rule_dict['sub_meta_rules'][_rule],
+ _rule_dict['rules'][_rule])
if _meta_rule_dict['aggregation'] == 'all_true':
return aggr_all_true(_decision_buffer)
return False
- def __get_key_from_value(self, value, values_dict):
- return filter(lambda v: v[1] == value, values_dict.iteritems())[0][0]
-
def get_intra_extension_list(self):
# TODO: check will be done through super_extension later
return self.driver.get_intra_extension_list()
@@ -404,20 +416,20 @@ class IntraExtensionManager(manager.Manager):
# We suppose that all subjects can be mapped to a true user in Keystone
for _subject in json_perimeter['subjects']:
user = self.identity_api.get_user_by_name(_subject, "default")
- subject_dict[user["id"]] = user["name"]
+ subject_dict[user["id"]] = user
self.driver.set_subject_dict(ie["id"], subject_dict)
ie["subjects"] = subject_dict
# Copy all values for objects and subjects
object_dict = dict()
for _object in json_perimeter['objects']:
- object_dict[uuid4().hex] = _object
+ object_dict[uuid4().hex] = {"name": _object}
self.driver.set_object_dict(ie["id"], object_dict)
ie["objects"] = object_dict
action_dict = dict()
for _action in json_perimeter['actions']:
- action_dict[uuid4().hex] = _action
+ action_dict[uuid4().hex] = {"name": _action}
self.driver.set_action_dict(ie["id"], action_dict)
ie["ations"] = action_dict
@@ -429,7 +441,7 @@ class IntraExtensionManager(manager.Manager):
subject_categories_dict = dict()
for _cat in json_perimeter['subject_categories']:
- subject_categories_dict[uuid4().hex] = _cat
+ subject_categories_dict[uuid4().hex] = {"name": _cat}
self.driver.set_subject_category_dict(ie["id"], subject_categories_dict)
# Initialize scope categories
for _cat in subject_categories_dict.keys():
@@ -438,7 +450,7 @@ class IntraExtensionManager(manager.Manager):
object_categories_dict = dict()
for _cat in json_perimeter['object_categories']:
- object_categories_dict[uuid4().hex] = _cat
+ object_categories_dict[uuid4().hex] = {"name": _cat}
self.driver.set_object_category_dict(ie["id"], object_categories_dict)
# Initialize scope categories
for _cat in object_categories_dict.keys():
@@ -447,7 +459,7 @@ class IntraExtensionManager(manager.Manager):
action_categories_dict = dict()
for _cat in json_perimeter['action_categories']:
- action_categories_dict[uuid4().hex] = _cat
+ action_categories_dict[uuid4().hex] = {"name": _cat}
self.driver.set_action_category_dict(ie["id"], action_categories_dict)
# Initialize scope categories
for _cat in action_categories_dict.keys():
@@ -462,34 +474,28 @@ class IntraExtensionManager(manager.Manager):
ie['subject_category_scope'] = dict()
for category, scope in json_perimeter["subject_category_scope"].iteritems():
- category = self.__get_key_from_value(
- category,
- self.driver.get_subject_category_dict(ie["id"])["subject_categories"])
+ category = self.driver.get_uuid_from_name(ie["id"], category, self.driver.SUBJECT_CATEGORY)
_scope_dict = dict()
for _scope in scope:
- _scope_dict[uuid4().hex] = _scope
+ _scope_dict[uuid4().hex] = {"name": _scope}
self.driver.set_subject_category_scope_dict(ie["id"], category, _scope_dict)
ie['subject_category_scope'][category] = _scope_dict
ie['object_category_scope'] = dict()
for category, scope in json_perimeter["object_category_scope"].iteritems():
- category = self.__get_key_from_value(
- category,
- self.driver.get_object_category_dict(ie["id"])["object_categories"])
+ category = self.driver.get_uuid_from_name(ie["id"], category, self.driver.OBJECT_CATEGORY)
_scope_dict = dict()
for _scope in scope:
- _scope_dict[uuid4().hex] = _scope
+ _scope_dict[uuid4().hex] = {"name": _scope}
self.driver.set_object_category_scope_dict(ie["id"], category, _scope_dict)
ie['object_category_scope'][category] = _scope_dict
ie['action_category_scope'] = dict()
for category, scope in json_perimeter["action_category_scope"].iteritems():
- category = self.__get_key_from_value(
- category,
- self.driver.get_action_category_dict(ie["id"])["action_categories"])
+ category = self.driver.get_uuid_from_name(ie["id"], category, self.driver.ACTION_CATEGORY)
_scope_dict = dict()
for _scope in scope:
- _scope_dict[uuid4().hex] = _scope
+ _scope_dict[uuid4().hex] = {"name": _scope}
self.driver.set_action_category_scope_dict(ie["id"], category, _scope_dict)
ie['action_category_scope'][category] = _scope_dict
@@ -499,18 +505,20 @@ class IntraExtensionManager(manager.Manager):
json_assignments = json.load(f)
subject_assignments = dict()
- for category, value in json_assignments['subject_assignments'].iteritems():
- category = self.__get_key_from_value(
- category,
- self.driver.get_subject_category_dict(ie["id"])["subject_categories"])
- for user in value:
- if user not in subject_assignments:
- subject_assignments[user] = dict()
- subject_assignments[user][category] = \
- map(lambda x: self.__get_key_from_value(x, ie['subject_category_scope'][category]), value[user])
+ for category_name, value in json_assignments['subject_assignments'].iteritems():
+ category = self.driver.get_uuid_from_name(ie["id"], category_name, self.driver.SUBJECT_CATEGORY)
+ for user_name in value:
+ user_uuid = self.driver.get_uuid_from_name(ie["id"], user_name, self.driver.SUBJECT)
+ if user_uuid not in subject_assignments:
+ subject_assignments[user_uuid] = dict()
+ if category not in subject_assignments[user_uuid]:
+ subject_assignments[user_uuid][category] = \
+ map(lambda x: self.driver.get_uuid_from_name(ie["id"], x, self.driver.SUBJECT_SCOPE, category_name),
+ value[user_name])
else:
- subject_assignments[user][category].extend(
- map(lambda x: self.__get_key_from_value(x, ie['subject_category_scope'][category]), value[user])
+ subject_assignments[user_uuid][category].extend(
+ map(lambda x: self.driver.get_uuid_from_name(ie["id"], x, self.driver.SUBJECT_SCOPE, category_name),
+ value[user_name])
)
# Note (dthom): subject_category_assignment must be initialized because when there is no data in json
# we will not go through the for loop
@@ -519,19 +527,18 @@ class IntraExtensionManager(manager.Manager):
self.driver.set_subject_category_assignment_dict(ie["id"], subject, subject_assignments[subject])
object_assignments = dict()
- for category, value in json_assignments["object_assignments"].iteritems():
- category = self.__get_key_from_value(
- category,
- self.driver.get_object_category_dict(ie["id"])["object_categories"])
+ for category_name, value in json_assignments["object_assignments"].iteritems():
+ category = self.driver.get_uuid_from_name(ie["id"], category_name, self.driver.OBJECT_CATEGORY)
for object_name in value:
if object_name not in object_assignments:
object_assignments[object_name] = dict()
+ if category not in object_assignments[object_name]:
object_assignments[object_name][category] = \
- map(lambda x: self.__get_key_from_value(x, ie['object_category_scope'][category]),
+ map(lambda x: self.driver.get_uuid_from_name(ie["id"], x, self.driver.OBJECT_SCOPE, category_name),
value[object_name])
else:
object_assignments[object_name][category].extend(
- map(lambda x: self.__get_key_from_value(x, ie['object_category_scope'][category]),
+ map(lambda x: self.driver.get_uuid_from_name(ie["id"], x, self.driver.OBJECT_SCOPE, category_name),
value[object_name])
)
# Note (dthom): object_category_assignment must be initialized because when there is no data in json
@@ -541,19 +548,18 @@ class IntraExtensionManager(manager.Manager):
self.driver.set_object_category_assignment_dict(ie["id"], object, object_assignments[object])
action_assignments = dict()
- for category, value in json_assignments["action_assignments"].iteritems():
- category = self.__get_key_from_value(
- category,
- self.driver.get_action_category_dict(ie["id"])["action_categories"])
+ for category_name, value in json_assignments["action_assignments"].iteritems():
+ category = self.driver.get_uuid_from_name(ie["id"], category_name, self.driver.ACTION_CATEGORY)
for action_name in value:
if action_name not in action_assignments:
action_assignments[action_name] = dict()
+ if category not in action_assignments[action_name]:
action_assignments[action_name][category] = \
- map(lambda x: self.__get_key_from_value(x, ie['action_category_scope'][category]),
+ map(lambda x: self.driver.get_uuid_from_name(ie["id"], x, self.driver.ACTION_SCOPE, category_name),
value[action_name])
else:
action_assignments[action_name][category].extend(
- map(lambda x: self.__get_key_from_value(x, ie['action_category_scope'][category]),
+ map(lambda x: self.driver.get_uuid_from_name(ie["id"], x, self.driver.ACTION_SCOPE, category_name),
value[action_name])
)
# Note (dthom): action_category_assignment must be initialized because when there is no data in json
@@ -570,9 +576,9 @@ class IntraExtensionManager(manager.Manager):
# ie["meta_rules"] = copy.deepcopy(json_metarule)
metarule = dict()
categories = {
- "subject_categories": self.driver.get_subject_category_dict(ie["id"]),
- "object_categories": self.driver.get_object_category_dict(ie["id"]),
- "action_categories": self.driver.get_action_category_dict(ie["id"])
+ "subject_categories": self.driver.SUBJECT_CATEGORY,
+ "object_categories": self.driver.OBJECT_CATEGORY,
+ "action_categories": self.driver.ACTION_CATEGORY
}
# Translate value from JSON file to UUID for Database
for relation in json_metarule["sub_meta_rules"]:
@@ -580,10 +586,8 @@ class IntraExtensionManager(manager.Manager):
for item in ("subject_categories", "object_categories", "action_categories"):
metarule[relation][item] = list()
for element in json_metarule["sub_meta_rules"][relation][item]:
- metarule[relation][item].append(self.__get_key_from_value(
- element,
- categories[item][item]
- ))
+ metarule[relation][item].append(self.driver.get_uuid_from_name(ie["id"], element, categories[item]))
+ metarule[relation]["algorithm"] = json_metarule["sub_meta_rules"][relation]["algorithm"]
submetarules = {
"aggregation": json_metarule["aggregation"],
"sub_meta_rules": metarule
@@ -600,33 +604,58 @@ class IntraExtensionManager(manager.Manager):
rules = dict()
sub_meta_rules = self.driver.get_meta_rule_dict(ie["id"])
for relation in json_rules:
- if relation not in self.get_sub_meta_rule_relations("admin", ie["id"])["sub_meta_rule_relations"]:
- raise IntraExtensionError("Bad relation name {} in rules".format(relation))
+ # print(relation)
+ # print(self.get_sub_meta_rule_relations("admin", ie["id"]))
+ # if relation not in self.get_sub_meta_rule_relations("admin", ie["id"])["sub_meta_rule_relations"]:
+ # raise IntraExtensionException("Bad relation name {} in rules".format(relation))
rules[relation] = list()
for rule in json_rules[relation]:
subrule = list()
- for cat, cat_func in (
- ("subject_categories", self.driver.get_subject_category_scope_dict),
- ("action_categories", self.driver.get_action_category_scope_dict),
- ("object_categories", self.driver.get_object_category_scope_dict),
- ):
- for cat_value in sub_meta_rules["sub_meta_rules"][relation][cat]:
- scope = cat_func(
- ie["id"],
- cat_value
- )[cat_func.__name__.replace("get_", "").replace("_dict", "")]
-
- _ = rule.pop(0)
- a_scope = self.__get_key_from_value(_, scope[cat_value])
- subrule.append(a_scope)
- # if a positive/negative value exists, all titem of rule have not be consumed
+ _rule = list(rule)
+ for category_uuid in sub_meta_rules["sub_meta_rules"][relation]["subject_categories"]:
+ scope_name = _rule.pop(0)
+ scope_uuid = self.driver.get_uuid_from_name(ie["id"],
+ scope_name,
+ self.driver.SUBJECT_SCOPE,
+ category_uuid=category_uuid)
+ subrule.append(scope_uuid)
+ for category_uuid in sub_meta_rules["sub_meta_rules"][relation]["action_categories"]:
+ scope_name = _rule.pop(0)
+ scope_uuid = self.driver.get_uuid_from_name(ie["id"],
+ scope_name,
+ self.driver.ACTION_SCOPE,
+ category_uuid=category_uuid)
+ subrule.append(scope_uuid)
+ for category_uuid in sub_meta_rules["sub_meta_rules"][relation]["object_categories"]:
+ scope_name = _rule.pop(0)
+ scope_uuid = self.driver.get_uuid_from_name(ie["id"],
+ scope_name,
+ self.driver.OBJECT_SCOPE,
+ category_uuid=category_uuid)
+ subrule.append(scope_uuid)
+ # for cat, cat_func, cat_func_cat in (
+ # ("subject_categories", self.driver.get_uuid_from_name, self.driver.SUBJECT_SCOPE),
+ # ("action_categories", self.driver.ACTION_SCOPE),
+ # ("object_categories", self.driver.OBJECT_SCOPE),
+ # ):
+ # for cat_value in sub_meta_rules["sub_meta_rules"][relation][cat]:
+ # scope = cat_func(
+ # ie["id"],
+ # cat_value,
+ # cat_func_cat
+ # )[cat_func.__name__.replace("get_", "").replace("_dict", "")]
+ #
+ # _ = rule.pop(0)
+ # a_scope = self.driver.get_uuid_from_name(ie["id"], _, scope[cat_value])
+ # subrule.append(a_scope)
+ # if a positive/negative value exists, all item of rule have not be consumed
if len(rule) >= 1 and type(rule[0]) is bool:
subrule.append(rule[0])
else:
# if value doesn't exist add a default value
subrule.append(True)
rules[relation].append(subrule)
- self.driver.set_rules(ie["id"], rules)
+ self.driver.set_rule_dict(ie["id"], rules)
def load_intra_extension(self, intra_extension):
ie = dict()
@@ -660,20 +689,26 @@ class IntraExtensionManager(manager.Manager):
@filter_args
@enforce(("read", "write"), "subjects")
- def set_subject_dict(self, user_uuid, intra_extension_uuid, subject_dict):
- for uuid in subject_dict:
+ def set_subject_dict(self, user_uuid, intra_extension_uuid, subject_list):
+ subject_dict = {}
+ for _user in subject_list:
# Next line will raise an error if user is not present in Keystone database
- self.identity_api.get_user(uuid)
+ user = self.identity_api.get_user_by_name(_user["name"], "default")
+ subject_dict[user["id"]] = dict()
+ for key in user.keys():
+ subject_dict[user["id"]][key] = user[key]
return self.driver.set_subject_dict(intra_extension_uuid, subject_dict)
@filter_args
@enforce(("read", "write"), "subjects")
- def add_subject_dict(self, user_uuid, intra_extension_uuid, subject_uuid):
+ def add_subject_dict(self, user_uuid, intra_extension_uuid, subject_name):
# Next line will raise an error if user is not present in Keystone database
- user = self.identity_api.get_user(subject_uuid)
- return self.driver.add_subject(intra_extension_uuid, subject_uuid, user["name"])
+ user = self.identity_api.get_user_by_name(subject_name, "default")
+ subjects = self.driver.add_subject(intra_extension_uuid, user["id"], user)
+ return subjects
@filter_args
+ @enforce("read", "subjects")
@enforce("write", "subjects")
def del_subject(self, user_uuid, intra_extension_uuid, subject_uuid):
self.driver.remove_subject(intra_extension_uuid, subject_uuid)
@@ -685,16 +720,20 @@ class IntraExtensionManager(manager.Manager):
@filter_args
@enforce(("read", "write"), "objects")
- def set_object_dict(self, user_uuid, intra_extension_uuid, object_dict):
+ def set_object_dict(self, user_uuid, intra_extension_uuid, object_list):
+ # TODO (asteroide): we must check input here.
+ object_dict = {uuid4().hex: item for item in object_list}
return self.driver.set_object_dict(intra_extension_uuid, object_dict)
@filter_args
@enforce(("read", "write"), "objects")
- def add_object_dict(self, user_uuid, intra_extension_uuid, object_name):
+ def add_object_dict(self, user_uuid, intra_extension_uuid, object_dict):
+ # TODO (asteroide): we must check input here.
object_uuid = uuid4().hex
- return self.driver.add_object(intra_extension_uuid, object_uuid, object_name)
+ return self.driver.add_object(intra_extension_uuid, object_uuid, object_dict)
@filter_args
+ @enforce("read", "objects")
@enforce("write", "objects")
def del_object(self, user_uuid, intra_extension_uuid, object_uuid):
self.driver.remove_object(intra_extension_uuid, object_uuid)
@@ -706,16 +745,20 @@ class IntraExtensionManager(manager.Manager):
@filter_args
@enforce(("read", "write"), "actions")
- def set_action_dict(self, user_uuid, intra_extension_uuid, action_dict):
+ def set_action_dict(self, user_uuid, intra_extension_uuid, action_list):
+ # TODO (asteroide): we must check input here.
+ action_dict = {uuid4().hex: item for item in action_list}
return self.driver.set_action_dict(intra_extension_uuid, action_dict)
@filter_args
@enforce(("read", "write"), "actions")
- def add_action_dict(self, user_uuid, intra_extension_uuid, action_name):
+ def add_action_dict(self, user_uuid, intra_extension_uuid, action_dict):
+ # TODO (asteroide): we must check input here.
action_uuid = uuid4().hex
- return self.driver.add_action(intra_extension_uuid, action_uuid, action_name)
+ return self.driver.add_action(intra_extension_uuid, action_uuid, action_dict)
@filter_args
+ @enforce("read", "actions")
@enforce("write", "actions")
def del_action(self, user_uuid, intra_extension_uuid, action_uuid):
self.driver.remove_action(intra_extension_uuid, action_uuid)
@@ -731,22 +774,26 @@ class IntraExtensionManager(manager.Manager):
@enforce("read", "subject_categories")
@enforce("read", "subject_category_scope")
@enforce("write", "subject_category_scope")
- def set_subject_category_dict(self, user_uuid, intra_extension_uuid, subject_category):
+ def set_subject_category_dict(self, user_uuid, intra_extension_uuid, subject_category_list):
+ subject_category = {uuid4().hex: item for item in subject_category_list}
subject_category_dict = self.driver.set_subject_category_dict(intra_extension_uuid, subject_category)
# if we add a new category, we must add it to the subject_category_scope
for _cat in subject_category.keys():
try:
_ = self.driver.get_subject_category_scope_dict(intra_extension_uuid, _cat)
- except AuthzMetadata:
+ except SubjectScopeUnknown:
self.driver.set_subject_category_scope_dict(intra_extension_uuid, _cat, {})
return subject_category_dict
@filter_args
@enforce("read", "subject_categories")
@enforce("write", "subject_categories")
- def add_subject_category_dict(self, user_uuid, intra_extension_uuid, subject_category_name):
+ def add_subject_category_dict(self, user_uuid, intra_extension_uuid, subject_category_dict):
+ # TODO (asteroide): we must check input here.
subject_category_uuid = uuid4().hex
- return self.driver.add_subject_category_dict(intra_extension_uuid, subject_category_uuid, subject_category_name)
+ subject_categories = self.driver.add_subject_category_dict(
+ intra_extension_uuid, subject_category_uuid, subject_category_dict)
+ return subject_categories
@filter_args
@enforce("write", "subject_categories")
@@ -762,22 +809,27 @@ class IntraExtensionManager(manager.Manager):
@enforce("read", "object_categories")
@enforce("read", "object_category_scope")
@enforce("write", "object_category_scope")
- def set_object_category_dict(self, user_uuid, intra_extension_uuid, object_category):
+ def set_object_category_dict(self, user_uuid, intra_extension_uuid, object_category_list):
+ # TODO (asteroide): we must check input here.
+ object_category = {uuid4().hex: item for item in object_category_list}
object_category_dict = self.driver.set_object_category_dict(intra_extension_uuid, object_category)
# if we add a new category, we must add it to the object_category_scope
for _cat in object_category.keys():
try:
_ = self.driver.get_object_category_scope_dict(intra_extension_uuid, _cat)
- except AuthzMetadata:
+ except ObjectScopeUnknown:
self.driver.set_object_category_scope_dict(intra_extension_uuid, _cat, {})
return object_category_dict
@filter_args
@enforce("read", "object_categories")
@enforce("write", "object_categories")
- def add_object_category_dict(self, user_uuid, intra_extension_uuid, object_category_name):
+ def add_object_category_dict(self, user_uuid, intra_extension_uuid, object_category_dict):
+ # TODO (asteroide): we must check input here.
object_category_uuid = uuid4().hex
- return self.driver.add_object_category_dict(intra_extension_uuid, object_category_uuid, object_category_name)
+ object_categories = self.driver.add_object_category_dict(
+ intra_extension_uuid, object_category_uuid, object_category_dict)
+ return object_categories
@filter_args
@enforce("write", "object_categories")
@@ -793,22 +845,27 @@ class IntraExtensionManager(manager.Manager):
@enforce("read", "action_categories")
@enforce("read", "action_category_scope")
@enforce("write", "action_category_scope")
- def set_action_category_dict(self, user_uuid, intra_extension_uuid, action_category):
+ def set_action_category_dict(self, user_uuid, intra_extension_uuid, action_category_list):
+ # TODO (asteroide): we must check input here.
+ action_category = {uuid4().hex: item for item in action_category_list}
action_category_dict = self.driver.set_action_category_dict(intra_extension_uuid, action_category)
# if we add a new category, we must add it to the action_category_scope
for _cat in action_category.keys():
try:
_ = self.driver.get_action_category_scope_dict(intra_extension_uuid, _cat)
- except AuthzMetadata:
+ except ActionScopeUnknown:
self.driver.set_action_category_scope_dict(intra_extension_uuid, _cat, {})
return action_category_dict
@filter_args
@enforce("read", "action_categories")
@enforce("write", "action_categories")
- def add_action_category_dict(self, user_uuid, intra_extension_uuid, action_category_name):
+ def add_action_category_dict(self, user_uuid, intra_extension_uuid, action_category_dict):
+ # TODO (asteroide): we must check input here.
action_category_uuid = uuid4().hex
- return self.driver.add_action_category_dict(intra_extension_uuid, action_category_uuid, action_category_name)
+ action_categories = self.driver.add_action_category_dict(
+ intra_extension_uuid, action_category_uuid, action_category_dict)
+ return action_categories
@filter_args
@enforce("write", "action_categories")
@@ -821,132 +878,107 @@ class IntraExtensionManager(manager.Manager):
@enforce("read", "subject_category")
def get_subject_category_scope_dict(self, user_uuid, intra_extension_uuid, category):
if category not in self.get_subject_category_dict(user_uuid, intra_extension_uuid)["subject_categories"]:
- raise IntraExtensionError("Subject category {} is unknown.".format(category))
+ raise SubjectCategoryUnknown()
return self.driver.get_subject_category_scope_dict(intra_extension_uuid, category)
@filter_args
@enforce("read", "subject_category_scope")
@enforce("read", "subject_category")
- def set_subject_category_scope_dict(self, user_uuid, intra_extension_uuid, category, scope):
- if category not in self.get_subject_category_dict(user_uuid, intra_extension_uuid)["subject_categories"]:
- raise IntraExtensionError("Subject category {} is unknown.".format(category))
- return self.driver.set_subject_category_scope_dict(intra_extension_uuid, category, scope)
+ def set_subject_category_scope_dict(self, user_uuid, intra_extension_uuid, category_uuid, scope_list):
+ # TODO (asteroide): we must check input here.
+ scope_dict = {uuid4().hex: item for item in scope_list}
+ return self.driver.set_subject_category_scope_dict(intra_extension_uuid, category_uuid, scope_dict)
@filter_args
@enforce(("read", "write"), "subject_category_scope")
@enforce("read", "subject_category")
- def add_subject_category_scope_dict(self, user_uuid, intra_extension_uuid, subject_category, scope_name):
- subject_categories = self.get_subject_category_dict(user_uuid, intra_extension_uuid)
- # check if subject_category exists in database
- if subject_category not in subject_categories["subject_categories"]:
- raise IntraExtensionError("Subject category {} is unknown.".format(subject_category))
+ def add_subject_category_scope_dict(self, user_uuid, intra_extension_uuid, category_uuid, scope_dict):
+ # TODO (asteroide): we must check input here.
scope_uuid = uuid4().hex
return self.driver.add_subject_category_scope_dict(
intra_extension_uuid,
- subject_category,
+ category_uuid,
scope_uuid,
- scope_name)
+ scope_dict)
@filter_args
@enforce("write", "subject_category_scope")
@enforce("read", "subject_category")
- def del_subject_category_scope(self, user_uuid, intra_extension_uuid, subject_category, subject_category_scope):
- subject_categories = self.get_subject_category_dict(user_uuid, intra_extension_uuid)
- # check if subject_category exists in database
- if subject_category not in subject_categories["subject_categories"]:
- raise IntraExtensionError("Subject category {} is unknown.".format(subject_category))
- return self.driver.remove_subject_category_scope_dict(
+ def del_subject_category_scope(self, user_uuid, intra_extension_uuid, category_uuid, scope_uuid):
+ self.driver.remove_subject_category_scope_dict(
intra_extension_uuid,
- subject_category,
- subject_category_scope)
+ category_uuid,
+ scope_uuid)
@filter_args
@enforce("read", "object_category_scope")
@enforce("read", "object_category")
- def get_object_category_scope_dict(self, user_uuid, intra_extension_uuid, category):
- if category not in self.get_object_category_dict(user_uuid, intra_extension_uuid)["object_categories"]:
- raise IntraExtensionError("Object category {} is unknown.".format(category))
- return self.driver.get_object_category_scope_dict(intra_extension_uuid, category)
+ def get_object_category_scope_dict(self, user_uuid, intra_extension_uuid, category_uuid):
+ return self.driver.get_object_category_scope_dict(intra_extension_uuid, category_uuid)
@filter_args
@enforce("read", "object_category_scope")
@enforce("read", "object_category")
- def set_object_category_scope_dict(self, user_uuid, intra_extension_uuid, category, scope):
- if category not in self.get_object_category_dict(user_uuid, intra_extension_uuid)["object_categories"]:
- raise IntraExtensionError("Object category {} is unknown.".format(category))
- return self.driver.set_object_category_scope_dict(intra_extension_uuid, category, scope)
+ def set_object_category_scope_dict(self, user_uuid, intra_extension_uuid, category_uuid, scope_list):
+ # TODO (asteroide): we must check input here.
+ scope_dict = {uuid4().hex: item for item in scope_list}
+ return self.driver.set_object_category_scope_dict(intra_extension_uuid, category_uuid, scope_dict)
@filter_args
@enforce(("read", "write"), "object_category_scope")
@enforce("read", "object_category")
- def add_object_category_scope_dict(self, user_uuid, intra_extension_uuid, object_category, scope_name):
- object_categories = self.get_object_category_dict(user_uuid, intra_extension_uuid)
- # check if object_category exists in database
- if object_category not in object_categories["object_categories"]:
- raise IntraExtensionError("Object category {} is unknown.".format(object_category))
+ def add_object_category_scope_dict(self, user_uuid, intra_extension_uuid, category_uuid, scope_dict):
+ # TODO (asteroide): we must check input here.
scope_uuid = uuid4().hex
- return self.driver.add_object_category_scope_dict(
+ return self.driver.add_subject_category_scope_dict(
intra_extension_uuid,
- object_category,
+ category_uuid,
scope_uuid,
- scope_name)
+ scope_dict)
@filter_args
@enforce("write", "object_category_scope")
@enforce("read", "object_category")
- def del_object_category_scope(self, user_uuid, intra_extension_uuid, object_category, object_category_scope):
- object_categories = self.get_object_category_dict(user_uuid, intra_extension_uuid)
- # check if object_category exists in database
- if object_category not in object_categories["object_categories"]:
- raise IntraExtensionError("Object category {} is unknown.".format(object_category))
- return self.driver.remove_object_category_scope_dict(
+ def del_object_category_scope(self, user_uuid, intra_extension_uuid, category_uuid, scope_uuid):
+ self.driver.remove_object_category_scope_dict(
intra_extension_uuid,
- object_category,
- object_category_scope)
+ category_uuid,
+ scope_uuid)
@filter_args
@enforce("read", "action_category_scope")
@enforce("read", "action_category")
- def get_action_category_scope_dict(self, user_uuid, intra_extension_uuid, category):
- if category not in self.get_action_category_dict(user_uuid, intra_extension_uuid)["action_categories"]:
- raise IntraExtensionError("Action category {} is unknown.".format(category))
- return self.driver.get_action_category_scope_dict(intra_extension_uuid, category)
+ def get_action_category_scope_dict(self, user_uuid, intra_extension_uuid, category_uuid):
+ return self.driver.get_action_category_scope_dict(intra_extension_uuid, category_uuid)
@filter_args
@enforce(("read", "write"), "action_category_scope")
@enforce("read", "action_category")
- def set_action_category_scope_dict(self, user_uuid, intra_extension_uuid, category, scope):
- if category not in self.get_action_category_dict(user_uuid, intra_extension_uuid)["action_categories"]:
- raise IntraExtensionError("Action category {} is unknown.".format(category))
- return self.driver.set_action_category_scope_dict(intra_extension_uuid, category, scope)
+ def set_action_category_scope_dict(self, user_uuid, intra_extension_uuid, category_uuid, scope_list):
+ # TODO (asteroide): we must check input here.
+ scope_dict = {uuid4().hex: item for item in scope_list}
+ return self.driver.set_action_category_scope_dict(intra_extension_uuid, category_uuid, scope_dict)
@filter_args
@enforce(("read", "write"), "action_category_scope")
@enforce("read", "action_category")
- def add_action_category_scope_dict(self, user_uuid, intra_extension_uuid, action_category, scope_name):
- action_categories = self.get_action_category_dict(user_uuid, intra_extension_uuid)
- # check if action_category exists in database
- if action_category not in action_categories["action_categories"]:
- raise IntraExtensionError("Action category {} is unknown.".format(action_category))
+ def add_action_category_scope_dict(self, user_uuid, intra_extension_uuid, category_uuid, scope_dict):
+ # TODO (asteroide): we must check input here.
scope_uuid = uuid4().hex
return self.driver.add_action_category_scope_dict(
intra_extension_uuid,
- action_category,
+ category_uuid,
scope_uuid,
- scope_name)
+ scope_dict)
@filter_args
@enforce("write", "action_category_scope")
@enforce("read", "action_category")
- def del_action_category_scope(self, user_uuid, intra_extension_uuid, action_category, action_category_scope):
- action_categories = self.get_action_category_dict(user_uuid, intra_extension_uuid)
- # check if action_category exists in database
- if action_category not in action_categories["action_categories"]:
- raise IntraExtensionError("Action category {} is unknown.".format(action_category))
- return self.driver.remove_action_category_scope_dict(
+ def del_action_category_scope(self, user_uuid, intra_extension_uuid, category_uuid, scope_uuid):
+ self.driver.remove_action_category_scope_dict(
intra_extension_uuid,
- action_category,
- action_category_scope)
+ category_uuid,
+ scope_uuid)
# Assignment functions
@@ -954,10 +986,6 @@ class IntraExtensionManager(manager.Manager):
@enforce("read", "subject_category_assignment")
@enforce("read", "subjects")
def get_subject_category_assignment_dict(self, user_uuid, intra_extension_uuid, subject_uuid):
- # check if subject exists in database
- if subject_uuid not in self.get_subject_dict(user_uuid, intra_extension_uuid)["subjects"]:
- LOG.error("add_subject_assignment: unknown subject_id {}".format(subject_uuid))
- raise IntraExtensionError("Bad input data")
return self.driver.get_subject_category_assignment_dict(intra_extension_uuid, subject_uuid)
@filter_args
@@ -965,10 +993,7 @@ class IntraExtensionManager(manager.Manager):
@enforce("write", "subject_category_assignment")
@enforce("read", "subjects")
def set_subject_category_assignment_dict(self, user_uuid, intra_extension_uuid, subject_uuid, assignment_dict):
- # check if subject exists in database
- if subject_uuid not in self.get_subject_dict(user_uuid, intra_extension_uuid)["subjects"]:
- LOG.error("add_subject_assignment: unknown subject_id {}".format(subject_uuid))
- raise IntraExtensionError("Bad input data")
+ # TODO (asteroide): we must check input here.
return self.driver.set_subject_category_assignment_dict(intra_extension_uuid, subject_uuid, assignment_dict)
@filter_args
@@ -976,40 +1001,27 @@ class IntraExtensionManager(manager.Manager):
@enforce("write", "subject_category_assignment")
@enforce("read", "subjects")
@enforce("read", "subject_category")
+ @enforce("read", "subject_scope")
def del_subject_category_assignment(self, user_uuid, intra_extension_uuid, subject_uuid, category_uuid, scope_uuid):
- # check if category exists in database
- if category_uuid not in self.get_subject_category_dict(user_uuid, intra_extension_uuid)["subject_categories"]:
- LOG.error("add_subject_category_scope: unknown subject_category {}".format(category_uuid))
- raise IntraExtensionError("Bad input data")
- # check if subject exists in database
- if subject_uuid not in self.get_subject_dict(user_uuid, intra_extension_uuid)["subjects"]:
- LOG.error("add_subject_assignment: unknown subject_id {}".format(subject_uuid))
- raise IntraExtensionError("Bad input data")
self.driver.remove_subject_category_assignment(intra_extension_uuid, subject_uuid, category_uuid, scope_uuid)
@filter_args
@enforce("write", "subject_category_assignment")
@enforce("read", "subjects")
@enforce("read", "subject_category")
- def add_subject_category_assignment_dict(self, user_uuid, intra_extension_uuid, subject_uuid, category_uuid, scope_uuid):
- # check if category exists in database
- if category_uuid not in self.get_subject_category_dict(user_uuid, intra_extension_uuid)["subject_categories"]:
- LOG.error("add_subject_category_scope: unknown subject_category {}".format(category_uuid))
- raise IntraExtensionError("Bad input data")
- # check if subject exists in database
- if subject_uuid not in self.get_subject_dict(user_uuid, intra_extension_uuid)["subjects"]:
- LOG.error("add_subject_assignment: unknown subject_id {}".format(subject_uuid))
- raise IntraExtensionError("Bad input data")
- return self.driver.add_subject_category_assignment_dict(intra_extension_uuid, subject_uuid, category_uuid, scope_uuid)
+ def add_subject_category_assignment_dict(self,
+ user_uuid,
+ intra_extension_uuid,
+ subject_uuid,
+ category_uuid,
+ scope_uuid):
+ return self.driver.add_subject_category_assignment_dict(
+ intra_extension_uuid, subject_uuid, category_uuid, scope_uuid)
@filter_args
@enforce("read", "object_category_assignment")
@enforce("read", "objects")
def get_object_category_assignment_dict(self, user_uuid, intra_extension_uuid, object_uuid):
- # check if object exists in database
- if object_uuid not in self.get_object_dict(user_uuid, intra_extension_uuid)["objects"]:
- LOG.error("add_object_assignment: unknown object_id {}".format(object_uuid))
- raise IntraExtensionError("Bad input data")
return self.driver.get_object_category_assignment_dict(intra_extension_uuid, object_uuid)
@filter_args
@@ -1017,10 +1029,7 @@ class IntraExtensionManager(manager.Manager):
@enforce("write", "object_category_assignment")
@enforce("read", "objects")
def set_object_category_assignment_dict(self, user_uuid, intra_extension_uuid, object_uuid, assignment_dict):
- # check if object exists in database
- if object_uuid not in self.get_object_dict(user_uuid, intra_extension_uuid)["objects"]:
- LOG.error("add_object_assignment: unknown object_id {}".format(object_uuid))
- raise IntraExtensionError("Bad input data")
+ # TODO (asteroide): we must check input here.
return self.driver.set_object_category_assignment_dict(intra_extension_uuid, object_uuid, assignment_dict)
@filter_args
@@ -1029,39 +1038,25 @@ class IntraExtensionManager(manager.Manager):
@enforce("read", "objects")
@enforce("read", "object_category")
def del_object_category_assignment(self, user_uuid, intra_extension_uuid, object_uuid, category_uuid, scope_uuid):
- # check if category exists in database
- if category_uuid not in self.get_object_category_dict(user_uuid, intra_extension_uuid)["object_categories"]:
- LOG.error("add_object_category_scope: unknown object_category {}".format(category_uuid))
- raise IntraExtensionError("Bad input data")
- # check if object exists in database
- if object_uuid not in self.get_object_dict(user_uuid, intra_extension_uuid)["objects"]:
- LOG.error("add_object_assignment: unknown object_id {}".format(object_uuid))
- raise IntraExtensionError("Bad input data")
self.driver.remove_object_category_assignment(intra_extension_uuid, object_uuid, category_uuid, scope_uuid)
@filter_args
@enforce("write", "object_category_assignment")
@enforce("read", "objects")
@enforce("read", "object_category")
- def add_object_category_assignment_dict(self, user_uuid, intra_extension_uuid, object_uuid, category_uuid, scope_uuid):
- # check if category exists in database
- if category_uuid not in self.get_object_category_dict(user_uuid, intra_extension_uuid)["object_categories"]:
- LOG.error("add_object_category_scope: unknown object_category {}".format(category_uuid))
- raise IntraExtensionError("Bad input data")
- # check if object exists in database
- if object_uuid not in self.get_object_dict(user_uuid, intra_extension_uuid)["objects"]:
- LOG.error("add_object_assignment: unknown object_id {}".format(object_uuid))
- raise IntraExtensionError("Bad input data")
- return self.driver.add_object_category_assignment_dict(intra_extension_uuid, object_uuid, category_uuid, scope_uuid)
+ def add_object_category_assignment_dict(self,
+ user_uuid,
+ intra_extension_uuid,
+ object_uuid,
+ category_uuid,
+ scope_uuid):
+ return self.driver.add_object_category_assignment_dict(
+ intra_extension_uuid, object_uuid, category_uuid, scope_uuid)
@filter_args
@enforce("read", "action_category_assignment")
@enforce("read", "actions")
def get_action_category_assignment_dict(self, user_uuid, intra_extension_uuid, action_uuid):
- # check if action exists in database
- if action_uuid not in self.get_action_dict(user_uuid, intra_extension_uuid)["actions"]:
- LOG.error("add_action_assignment: unknown action_id {}".format(action_uuid))
- raise IntraExtensionError("Bad input data")
return self.driver.get_action_category_assignment_dict(intra_extension_uuid, action_uuid)
@filter_args
@@ -1069,10 +1064,7 @@ class IntraExtensionManager(manager.Manager):
@enforce("write", "action_category_assignment")
@enforce("read", "actions")
def set_action_category_assignment_dict(self, user_uuid, intra_extension_uuid, action_uuid, assignment_dict):
- # check if action exists in database
- if action_uuid not in self.get_action_dict(user_uuid, intra_extension_uuid)["actions"]:
- LOG.error("add_action_assignment: unknown action_id {}".format(action_uuid))
- raise IntraExtensionError("Bad input data")
+ # TODO (asteroide): we must check input here.
return self.driver.set_action_category_assignment_dict(intra_extension_uuid, action_uuid, assignment_dict)
@filter_args
@@ -1081,29 +1073,18 @@ class IntraExtensionManager(manager.Manager):
@enforce("read", "actions")
@enforce("read", "action_category")
def del_action_category_assignment(self, user_uuid, intra_extension_uuid, action_uuid, category_uuid, scope_uuid):
- # check if category exists in database
- if category_uuid not in self.get_action_category_dict(user_uuid, intra_extension_uuid)["action_categories"]:
- LOG.error("add_action_category_scope: unknown action_category {}".format(category_uuid))
- raise IntraExtensionError("Bad input data")
- # check if action exists in database
- if action_uuid not in self.get_action_dict(user_uuid, intra_extension_uuid)["actions"]:
- LOG.error("add_action_assignment: unknown action_id {}".format(action_uuid))
- raise IntraExtensionError("Bad input data")
self.driver.remove_action_category_assignment(intra_extension_uuid, action_uuid, category_uuid, scope_uuid)
@filter_args
@enforce("write", "action_category_assignment")
@enforce("read", "actions")
@enforce("read", "action_category")
- def add_action_category_assignment_dict(self, user_uuid, intra_extension_uuid, action_uuid, category_uuid, scope_uuid):
- # check if category exists in database
- if category_uuid not in self.get_action_category_dict(user_uuid, intra_extension_uuid)["action_categories"]:
- LOG.error("add_action_category_scope: unknown action_category {}".format(category_uuid))
- raise IntraExtensionError("Bad input data")
- # check if action exists in database
- if action_uuid not in self.get_action_dict(user_uuid, intra_extension_uuid)["actions"]:
- LOG.error("add_action_assignment: unknown action_id {}".format(action_uuid))
- raise IntraExtensionError("Bad input data")
+ def add_action_category_assignment_dict(self,
+ user_uuid,
+ intra_extension_uuid,
+ action_uuid,
+ category_uuid,
+ scope_uuid):
return self.driver.add_action_category_assignment_dict(
intra_extension_uuid,
action_uuid,
@@ -1115,7 +1096,7 @@ class IntraExtensionManager(manager.Manager):
@filter_args
def get_aggregation_algorithms(self, user_uuid, intra_extension_uuid):
# TODO: check which algorithms are really usable
- return {"aggregation_algorithms": ["and_true_aggregation", "test_aggregation"]}
+ return {"aggregation_algorithms": ["and_true_aggregation", "test_aggregation", "all_true"]}
@filter_args
@enforce("read", "aggregation_algorithms")
@@ -1142,37 +1123,32 @@ class IntraExtensionManager(manager.Manager):
@enforce("read", "sub_meta_rule")
@enforce("write", "sub_meta_rule")
def set_sub_meta_rule(self, user_uuid, intra_extension_uuid, sub_meta_rules):
+ # TODO: check which algorithms are really usable
# TODO (dthom): When sub_meta_rule is set, all rules must be dropped
# because the previous rules cannot be mapped to the new sub_meta_rule.
for relation in sub_meta_rules.keys():
if relation not in self.get_sub_meta_rule_relations(user_uuid, intra_extension_uuid)["sub_meta_rule_relations"]:
- LOG.error("set_sub_meta_rule unknown MetaRule relation {}".format(relation))
- raise IntraExtensionError("Bad input data.")
+ raise IntraExtensionError("set_sub_meta_rule unknown MetaRule relation {}".format(relation))
for cat in ("subject_categories", "object_categories", "action_categories"):
if cat not in sub_meta_rules[relation]:
- LOG.error("set_sub_meta_rule category {} missed".format(cat))
- raise IntraExtensionError("Bad input data.")
+ raise IntraExtensionError("set_sub_meta_rule category {} missed".format(cat))
if type(sub_meta_rules[relation][cat]) is not list:
- LOG.error("set_sub_meta_rule category {} is not a list".format(cat))
- raise IntraExtensionError("Bad input data.")
- subject_categories = self.get_subject_category_dict(user_uuid, intra_extension_uuid)
- for data in sub_meta_rules[relation]["subject_categories"]:
- if data not in subject_categories["subject_categories"]:
- LOG.error("set_sub_meta_rule category {} is not part of subject_categories {}".format(
- data, subject_categories))
- raise IntraExtensionError("Bad input data.")
- object_categories = self.get_object_category_dict(user_uuid, intra_extension_uuid)
+ raise IntraExtensionError("set_sub_meta_rule category {} is not a list".format(cat))
+ subject_categories = self.get_subject_category_dict(user_uuid, intra_extension_uuid)["subject_categories"]
+ for data in list(sub_meta_rules[relation]["subject_categories"]):
+ if data not in subject_categories:
+ raise IntraExtensionError("set_sub_meta_rule category {} is not part of subject_categories {}".format(
+ data, subject_categories))
+ object_categories = self.get_object_category_dict(user_uuid, intra_extension_uuid)["object_categories"]
for data in sub_meta_rules[relation]["object_categories"]:
- if data not in object_categories["object_categories"]:
- LOG.error("set_sub_meta_rule category {} is not part of object_categories {}".format(
- data, object_categories))
- raise IntraExtensionError("Bad input data.")
- action_categories = self.get_action_category_dict(user_uuid, intra_extension_uuid)
+ if data not in object_categories:
+ raise IntraExtensionError("set_sub_meta_rule category {} is not part of object_categories {}".format(
+ data, object_categories))
+ action_categories = self.get_action_category_dict(user_uuid, intra_extension_uuid)["actions_categories"]
for data in sub_meta_rules[relation]["action_categories"]:
- if data not in action_categories["action_categories"]:
- LOG.error("set_sub_meta_rule category {} is not part of action_categories {}".format(
- data, action_categories))
- raise IntraExtensionError("Bad input data.")
+ if data not in action_categories:
+ raise IntraExtensionError("set_sub_meta_rule category {} is not part of action_categories {}".format(
+ data, action_categories))
aggregation = self.driver.get_meta_rule_dict(intra_extension_uuid)["aggregation"]
return self.driver.set_meta_rule_dict(
intra_extension_uuid,
@@ -1184,13 +1160,13 @@ class IntraExtensionManager(manager.Manager):
# Sub-rules functions
@filter_args
@enforce("read", "sub_rules")
- def get_sub_rules(self, user_uuid, intra_extension_uuid):
+ def get_rules(self, user_uuid, intra_extension_uuid):
return self.driver.get_rules(intra_extension_uuid)
@filter_args
@enforce("read", "sub_rules")
@enforce("write", "sub_rules")
- def set_sub_rule(self, user_uuid, intra_extension_uuid, relation, sub_rule):
+ def set_rule(self, user_uuid, intra_extension_uuid, relation, sub_rule):
for item in sub_rule:
if type(item) not in (str, unicode, bool):
raise IntraExtensionError("Bad input data (sub_rule).")
@@ -1239,22 +1215,22 @@ class IntraExtensionManager(manager.Manager):
@filter_args
@enforce("read", "sub_rules")
@enforce("write", "sub_rules")
- def del_sub_rule(self, user_uuid, intra_extension_uuid, relation_name, rule):
+ def del_rule(self, user_uuid, intra_extension_uuid, relation_name, sub_rule):
ref_rules = self.driver.get_rules(intra_extension_uuid)
- rule = rule.split("+")
- for index, _item in enumerate(rule):
+ sub_rule = sub_rule.split("+")
+ for index, _item in enumerate(sub_rule):
if "True" in _item:
- rule[index] = True
+ sub_rule[index] = True
if "False" in _item:
- rule[index] = False
+ sub_rule[index] = False
if relation_name in ref_rules["rules"]:
- if rule in ref_rules["rules"][relation_name]:
- ref_rules["rules"][relation_name].remove(rule)
+ if sub_rule in ref_rules["rules"][relation_name]:
+ ref_rules["rules"][relation_name].remove(sub_rule)
else:
- self.moonlog_api.error("Unknown rule: {}".format(rule))
+ raise RuleUnknown()
else:
- self.moonlog_api.error("Unknown relation name for rules: {}".format(relation_name))
- return self.driver.set_rules(intra_extension_uuid, ref_rules["rules"])
+ raise IntraExtensionError("Unknown relation name for rules: {}".format(relation_name))
+ self.driver.set_rules(intra_extension_uuid, ref_rules["rules"])
@dependency.provider('authz_api')
@@ -1281,10 +1257,10 @@ class IntraExtensionAuthzManager(IntraExtensionManager):
def set_subject_dict(self, user_uuid, intra_extension_uuid, subject_dict):
raise SubjectAddNotAuthorized()
- def add_subject_dict(self, user_uuid, intra_extension_uuid, subject_uuid):
+ def add_subject_dict(self, user_uuid, intra_extension_uuid, subject_name):
raise SubjectAddNotAuthorized()
- def del_subject(self, user_uuid, intra_extension_uuid, subject_uuid):
+ def del_subject(self, user_uuid, intra_extension_uuid, subject_name):
raise SubjectDelNotAuthorized()
def set_object_dict(self, user_uuid, intra_extension_uuid, object_dict):
@@ -1299,7 +1275,7 @@ class IntraExtensionAuthzManager(IntraExtensionManager):
def set_action_dict(self, user_uuid, intra_extension_uuid, action_dict):
raise ActionAddNotAuthorized()
- def add_action_dict(self, user_uuid, intra_extension_uuid, action_name):
+ def add_action_dict(self, user_uuid, intra_extension_uuid, action_dict):
raise ActionAddNotAuthorized()
def del_action(self, user_uuid, intra_extension_uuid, action_uuid):
@@ -1509,6 +1485,112 @@ class UpdateDriver(object):
class IntraExtensionDriver(object):
+ SUBJECT = 'subject'
+ OBJECT = 'object'
+ ACTION = 'action'
+ SUBJECT_CATEGORY = 'subject_category'
+ OBJECT_CATEGORY = 'object_category'
+ ACTION_CATEGORY = 'action_category'
+ SUBJECT_SCOPE = 'subject_scope'
+ OBJECT_SCOPE = 'object_scope'
+ ACTION_SCOPE = 'action_scope'
+
+ def __get_data_from_type(self,
+ intra_extension_uuid,
+ name=None,
+ uuid=None,
+ data_name=None,
+ category_name=None,
+ category_uuid=None):
+
+ def extract_name(data_dict):
+ for key in data_dict:
+ try:
+ yield data_dict[key]["name"]
+ except KeyError:
+ for key2 in data_dict[key]:
+ yield data_dict[key][key2]["name"]
+
+ data_values = list()
+
+ if data_name == self.SUBJECT:
+ data_values = self.get_subject_dict(intra_extension_uuid)["subjects"]
+ if (name and name not in extract_name(data_values)) or \
+ (uuid and uuid not in data_values.keys()):
+ raise SubjectUnknown()
+ elif data_name == self.OBJECT:
+ data_values = self.get_object_dict(intra_extension_uuid)["objects"]
+ if (name and name not in extract_name(data_values)) or \
+ (uuid and uuid not in data_values.keys()):
+ raise ObjectUnknown()
+ elif data_name == self.ACTION:
+ data_values = self.get_action_dict(intra_extension_uuid)["actions"]
+ if (name and name not in extract_name(data_values)) or \
+ (uuid and uuid not in data_values.keys()):
+ raise ActionUnknown()
+ elif data_name == self.SUBJECT_CATEGORY:
+ data_values = self.get_subject_category_dict(intra_extension_uuid)["subject_categories"]
+ if (name and name not in extract_name(data_values)) or \
+ (uuid and uuid not in data_values.keys()):
+ raise SubjectCategoryUnknown()
+ elif data_name == self.OBJECT_CATEGORY:
+ data_values = self.get_object_category_dict(intra_extension_uuid)["object_categories"]
+ if (name and name not in extract_name(data_values)) or \
+ (uuid and uuid not in data_values.keys()):
+ raise ObjectCategoryUnknown()
+ elif data_name == self.ACTION_CATEGORY:
+ data_values = self.get_action_category_dict(intra_extension_uuid)["action_categories"]
+ if (name and name not in extract_name(data_values)) or \
+ (uuid and uuid not in data_values.keys()):
+ raise ActionCategoryUnknown()
+ elif data_name == self.SUBJECT_SCOPE:
+ if not category_uuid:
+ category_uuid = self.get_uuid_from_name(intra_extension_uuid, category_name, self.SUBJECT_CATEGORY)
+ data_values = self.get_subject_category_scope_dict(intra_extension_uuid,
+ category_uuid)["subject_category_scope"]
+ if (name and name not in extract_name(data_values)) or \
+ (uuid and uuid not in data_values.keys()):
+ raise SubjectScopeUnknown()
+ elif data_name == self.OBJECT_SCOPE:
+ if not category_uuid:
+ category_uuid = self.get_uuid_from_name(intra_extension_uuid, category_name, self.OBJECT_CATEGORY)
+ data_values = self.get_object_category_scope_dict(intra_extension_uuid,
+ category_uuid)["object_category_scope"]
+ if (name and name not in extract_name(data_values)) or \
+ (uuid and uuid not in data_values.keys()):
+ raise ObjectScopeUnknown()
+ elif data_name == self.ACTION_SCOPE:
+ if not category_uuid:
+ category_uuid = self.get_uuid_from_name(intra_extension_uuid, category_name, self.ACTION_CATEGORY)
+ data_values = self.get_action_category_scope_dict(intra_extension_uuid,
+ category_uuid)["action_category_scope"]
+ if (name and name not in extract_name(data_values)) or \
+ (uuid and uuid not in data_values.keys()):
+ raise ActionScopeUnknown()
+ if category_uuid:
+ return data_values[category_uuid]
+ return data_values
+
+ def get_uuid_from_name(self, intra_extension_uuid, name, data_name, category_name=None, category_uuid=None):
+ data_values = self.__get_data_from_type(
+ intra_extension_uuid=intra_extension_uuid,
+ name=name,
+ data_name=data_name,
+ category_name=category_name,
+ category_uuid=category_uuid,
+ )
+ return filter(lambda v: v[1]["name"] == name, data_values.iteritems())[0][0]
+
+ def get_name_from_uuid(self, intra_extension_uuid, uuid, data_name, category_name=None, category_uuid=None):
+ data_values = self.__get_data_from_type(
+ intra_extension_uuid=intra_extension_uuid,
+ uuid=uuid,
+ data_name=data_name,
+ category_name=category_name,
+ category_uuid=category_uuid,
+ )
+ return data_values[uuid]
+
# Getter ad Setter for subjects
def get_subject_dict(self, extension_uuid):
diff --git a/keystone-moon/keystone/contrib/moon/exception.py b/keystone-moon/keystone/contrib/moon/exception.py
index 8ff77b7e..47b9c2f1 100644
--- a/keystone-moon/keystone/contrib/moon/exception.py
+++ b/keystone-moon/keystone/contrib/moon/exception.py
@@ -147,6 +147,48 @@ class ActionUnknown(AuthzPerimeter):
logger = "ERROR"
+class SubjectCategoryUnknown(AuthzMetadata):
+ message_format = _("The given subject category is unknown.")
+ code = 400
+ title = 'Subject Category Unknown'
+ logger = "ERROR"
+
+
+class ObjectCategoryUnknown(AuthzMetadata):
+ message_format = _("The given object category is unknown.")
+ code = 400
+ title = 'Object Category Unknown'
+ logger = "ERROR"
+
+
+class ActionCategoryUnknown(AuthzMetadata):
+ message_format = _("The given action category is unknown.")
+ code = 400
+ title = 'Action Category Unknown'
+ logger = "ERROR"
+
+
+class SubjectScopeUnknown(AuthzScope):
+ message_format = _("The given subject scope is unknown.")
+ code = 400
+ title = 'Subject Scope Unknown'
+ logger = "ERROR"
+
+
+class ObjectScopeUnknown(AuthzScope):
+ message_format = _("The given object scope is unknown.")
+ code = 400
+ title = 'Object Scope Unknown'
+ logger = "ERROR"
+
+
+class ActionScopeUnknown(AuthzScope):
+ message_format = _("The given action scope is unknown.")
+ code = 400
+ title = 'Action Scope Unknown'
+ logger = "ERROR"
+
+
class SubjectCategoryAssignmentOutOfScope(AuthzScope):
message_format = _("The given subject category scope value is out of scope.")
code = 400
diff --git a/keystone-moon/keystone/tests/moon/unit/test_unit_core_intra_extension_admin.py b/keystone-moon/keystone/tests/moon/unit/test_unit_core_intra_extension_admin.py
index 684b9695..f527ee9e 100644
--- a/keystone-moon/keystone/tests/moon/unit/test_unit_core_intra_extension_admin.py
+++ b/keystone-moon/keystone/tests/moon/unit/test_unit_core_intra_extension_admin.py
@@ -60,7 +60,7 @@ class TestIntraExtensionAdminManagerOK(tests.TestCase):
group='moon',
policy_directory=self.policy_directory)
- def create_intra_extension(self, policy_model="policy_rbac_admin"):
+ def create_intra_extension(self, policy_model="policy_admin"):
# Create the admin user because IntraExtension needs it
#self.admin = self.identity_api.create_user(USER)
IE["policymodel"] = policy_model
@@ -99,6 +99,7 @@ class TestIntraExtensionAdminManagerOK(tests.TestCase):
self.manager.delete_intra_extension(self.ref["id"])
def test_subjects(self):
+ self.create_user("demo")
self.create_user("admin")
self.create_intra_extension()
@@ -147,6 +148,7 @@ class TestIntraExtensionAdminManagerOK(tests.TestCase):
self.assertIn(new_subject["id"], subjects["subjects"])
def test_objects(self):
+ self.create_user("demo")
self.create_user("admin")
self.create_intra_extension()
@@ -196,6 +198,7 @@ class TestIntraExtensionAdminManagerOK(tests.TestCase):
self.assertIn(new_object["id"], objects["objects"])
def test_actions(self):
+ self.create_user("demo")
self.create_user("admin")
self.create_intra_extension()
@@ -245,6 +248,7 @@ class TestIntraExtensionAdminManagerOK(tests.TestCase):
self.assertIn(new_action["id"], actions["actions"])
def test_subject_categories(self):
+ self.create_user("demo")
self.create_user("admin")
self.create_intra_extension()
@@ -299,6 +303,7 @@ class TestIntraExtensionAdminManagerOK(tests.TestCase):
self.assertIn(new_subject_category["id"], subject_categories["subject_categories"])
def test_object_categories(self):
+ self.create_user("demo")
self.create_user("admin")
self.create_intra_extension()
@@ -353,6 +358,7 @@ class TestIntraExtensionAdminManagerOK(tests.TestCase):
self.assertIn(new_object_category["id"], object_categories["object_categories"])
def test_action_categories(self):
+ self.create_user("demo")
self.create_user("admin")
self.create_intra_extension()
@@ -407,6 +413,7 @@ class TestIntraExtensionAdminManagerOK(tests.TestCase):
self.assertIn(new_action_category["id"], action_categories["action_categories"])
def test_subject_category_scope(self):
+ self.create_user("demo")
self.create_user("admin")
self.create_intra_extension()
@@ -487,6 +494,7 @@ class TestIntraExtensionAdminManagerOK(tests.TestCase):
self.assertNotIn(new_subject_category_scope_uuid, subject_category_scope["subject_category_scope"])
def test_object_category_scope(self):
+ self.create_user("demo")
self.create_user("admin")
self.create_intra_extension()
@@ -567,6 +575,7 @@ class TestIntraExtensionAdminManagerOK(tests.TestCase):
self.assertNotIn(new_object_category_scope_uuid, object_category_scope["object_category_scope"])
def test_action_category_scope(self):
+ self.create_user("demo")
self.create_user("admin")
self.create_intra_extension()
@@ -647,6 +656,7 @@ class TestIntraExtensionAdminManagerOK(tests.TestCase):
self.assertNotIn(new_action_category_scope_uuid, action_category_scope["action_category_scope"])
def test_subject_category_assignment(self):
+ self.create_user("demo")
self.create_user("admin")
self.create_intra_extension()
@@ -795,6 +805,7 @@ class TestIntraExtensionAdminManagerOK(tests.TestCase):
subject_category_assignments["subject_category_assignments"][new_subject["id"]])
def test_object_category_assignment(self):
+ self.create_user("demo")
self.create_user("admin")
self.create_intra_extension()
@@ -943,6 +954,7 @@ class TestIntraExtensionAdminManagerOK(tests.TestCase):
object_category_assignments["object_category_assignments"][new_object["id"]])
def test_action_category_assignment(self):
+ self.create_user("demo")
self.create_user("admin")
self.create_intra_extension()
@@ -1091,6 +1103,7 @@ class TestIntraExtensionAdminManagerOK(tests.TestCase):
action_category_assignments["action_category_assignments"][new_action["id"]])
def test_sub_meta_rules(self):
+ self.create_user("demo")
self.create_user("admin")
self.create_intra_extension()
@@ -1166,6 +1179,7 @@ class TestIntraExtensionAdminManagerOK(tests.TestCase):
)
def test_sub_rules(self):
+ self.create_user("demo")
self.create_user("admin")
self.create_intra_extension()
@@ -1292,7 +1306,7 @@ class TestIntraExtensionAdminManagerKO(tests.TestCase):
_USER["name"] = username
return self.identity_api.create_user(_USER)
- def create_intra_extension(self, policy_model="policy_rbac_authz"):
+ def create_intra_extension(self, policy_model="policy_authz"):
IE["policymodel"] = policy_model
IE["name"] = uuid.uuid4().hex
@@ -1302,8 +1316,8 @@ class TestIntraExtensionAdminManagerKO(tests.TestCase):
def test_subjects(self):
admin_user = self.create_user("admin")
- ref = self.create_intra_extension()
demo_user = self.create_user("demo")
+ ref = self.create_intra_extension()
self.assertRaises(
SubjectReadNotAuthorized,
@@ -1372,8 +1386,8 @@ class TestIntraExtensionAdminManagerKO(tests.TestCase):
def test_objects(self):
admin_user = self.create_user("admin")
- ref = self.create_intra_extension()
demo_user = self.create_user("demo")
+ ref = self.create_intra_extension()
self.assertRaises(
ObjectReadNotAuthorized,
@@ -1443,8 +1457,8 @@ class TestIntraExtensionAdminManagerKO(tests.TestCase):
def test_actions(self):
admin_user = self.create_user("admin")
- ref = self.create_intra_extension()
demo_user = self.create_user("demo")
+ ref = self.create_intra_extension()
self.assertRaises(
ActionReadNotAuthorized,
@@ -1513,9 +1527,9 @@ class TestIntraExtensionAdminManagerKO(tests.TestCase):
self.assertIn(new_action["id"], actions["actions"])
def test_subject_categories(self):
+ demo_user = self.create_user("demo")
admin_user = self.create_user("admin")
ref = self.create_intra_extension()
- demo_user = self.create_user("demo")
self.assertRaises(
SubjectCategoryReadNotAuthorized,
@@ -1589,9 +1603,9 @@ class TestIntraExtensionAdminManagerKO(tests.TestCase):
self.assertIn(new_subject_category["id"], subject_categories["subject_categories"])
def test_object_categories(self):
+ demo_user = self.create_user("demo")
admin_user = self.create_user("admin")
ref = self.create_intra_extension()
- demo_user = self.create_user("demo")
self.assertRaises(
ObjectCategoryReadNotAuthorized,
@@ -1665,9 +1679,15 @@ class TestIntraExtensionAdminManagerKO(tests.TestCase):
self.assertIn(new_object_category["id"], object_categories["object_categories"])
def test_action_categories(self):
- admin_user = self.create_user("admin")
- ref = self.create_intra_extension()
+ admin_user = self.create_user()
demo_user = self.create_user("demo")
+ tenant = self.create_tenant()
+ ie_authz = self.create_intra_extension("policy_authz")
+ ie_admin = self.create_intra_extension("policy_admin")
+ mapping = self.create_mapping(tenant, ie_authz["id"], ie_admin["id"])
+ ref = ie_authz
+ # admin_user = self.create_user("admin")
+ # ref = self.create_intra_extension()
self.assertRaises(
ActionCategoryReadNotAuthorized,
@@ -1740,9 +1760,9 @@ class TestIntraExtensionAdminManagerKO(tests.TestCase):
self.assertIn(new_action_category["id"], action_categories["action_categories"])
def test_subject_category_scope(self):
+ demo_user = self.create_user("demo")
admin_user = self.create_user("admin")
ref = self.create_intra_extension()
- demo_user = self.create_user("demo")
subject_categories = self.manager.set_subject_category_dict(
admin_user["id"],
@@ -1843,9 +1863,9 @@ class TestIntraExtensionAdminManagerKO(tests.TestCase):
self.assertNotIn(new_subject_category_scope_uuid, subject_category_scope["subject_category_scope"])
def test_object_category_scope(self):
+ demo_user = self.create_user("demo")
admin_user = self.create_user("admin")
ref = self.create_intra_extension()
- demo_user = self.create_user("demo")
object_categories = self.manager.set_object_category_dict(
admin_user["id"],
@@ -1947,9 +1967,9 @@ class TestIntraExtensionAdminManagerKO(tests.TestCase):
self.assertNotIn(new_object_category_scope_uuid, object_category_scope["object_category_scope"])
def test_action_category_scope(self):
+ demo_user = self.create_user("demo")
admin_user = self.create_user("admin")
ref = self.create_intra_extension()
- demo_user = self.create_user("demo")
action_categories = self.manager.set_action_category_dict(
admin_user["id"],
@@ -2053,9 +2073,9 @@ class TestIntraExtensionAdminManagerKO(tests.TestCase):
self.assertNotIn(new_action_category_scope_uuid, action_category_scope["action_category_scope"])
def test_subject_category_assignment(self):
+ demo_user = self.create_user("demo")
admin_user = self.create_user("admin")
ref = self.create_intra_extension()
- demo_user = self.create_user("demo")
new_subject = self.create_user()
new_subjects = dict()
@@ -2232,9 +2252,9 @@ class TestIntraExtensionAdminManagerKO(tests.TestCase):
subject_category_assignments["subject_category_assignments"][new_subject["id"]])
def test_object_category_assignment(self):
+ demo_user = self.create_user("demo")
admin_user = self.create_user("admin")
ref = self.create_intra_extension()
- demo_user = self.create_user("demo")
new_object = {"id": uuid.uuid4().hex, "name": "my_object"}
new_objects = dict()
@@ -2412,9 +2432,9 @@ class TestIntraExtensionAdminManagerKO(tests.TestCase):
object_category_assignments["object_category_assignments"][new_object["id"]])
def test_action_category_assignment(self):
+ demo_user = self.create_user("demo")
admin_user = self.create_user("admin")
ref = self.create_intra_extension()
- demo_user = self.create_user("demo")
new_action = {"id": uuid.uuid4().hex, "name": "my_action"}
new_actions = dict()
@@ -2592,9 +2612,9 @@ class TestIntraExtensionAdminManagerKO(tests.TestCase):
action_category_assignments["action_category_assignments"][new_action["id"]])
def test_sub_meta_rules(self):
+ demo_user = self.create_user("demo")
admin_user = self.create_user("admin")
ref = self.create_intra_extension()
- demo_user = self.create_user("demo")
aggregation_algorithms = self.manager.get_aggregation_algorithms(admin_user["id"], ref["id"])
self.assertIsInstance(aggregation_algorithms, dict)
@@ -2694,9 +2714,9 @@ class TestIntraExtensionAdminManagerKO(tests.TestCase):
)
def test_sub_rules(self):
+ demo_user = self.create_user("demo")
admin_user = self.create_user("admin")
ref = self.create_intra_extension()
- demo_user = self.create_user("demo")
sub_meta_rules = self.manager.get_sub_meta_rule(admin_user["id"], ref["id"])
self.assertIsInstance(sub_meta_rules, dict)
diff --git a/keystone-moon/keystone/tests/moon/unit/test_unit_core_intra_extension_authz.py b/keystone-moon/keystone/tests/moon/unit/test_unit_core_intra_extension_authz.py
index 4752632b..e2e151ed 100644
--- a/keystone-moon/keystone/tests/moon/unit/test_unit_core_intra_extension_authz.py
+++ b/keystone-moon/keystone/tests/moon/unit/test_unit_core_intra_extension_authz.py
@@ -27,7 +27,7 @@ USER = {
IE = {
"name": "test IE",
- "policymodel": "policy_rbac_authz",
+ "policymodel": "policy_authz",
"description": "a simple description."
}
@@ -85,7 +85,7 @@ class TestIntraExtensionAuthzManagerAuthz(tests.TestCase):
_USER["name"] = username
return self.identity_api.create_user(_USER)
- def create_intra_extension(self, policy_model="policy_rbac_authz"):
+ def create_intra_extension(self, policy_model="policy_authz"):
IE["policymodel"] = policy_model
IE["name"] = uuid.uuid4().hex
@@ -149,8 +149,8 @@ class TestIntraExtensionAuthzManagerAuthz(tests.TestCase):
admin_user = self.create_user()
tenant = self.create_tenant()
- ie_authz = self.create_intra_extension("policy_rbac_authz")
- ie_admin = self.create_intra_extension("policy_rbac_admin")
+ ie_authz = self.create_intra_extension("policy_authz")
+ ie_admin = self.create_intra_extension("policy_admin")
mapping = self.create_mapping(tenant, ie_authz["id"], ie_admin["id"])
# Test when subject is unknown
@@ -338,10 +338,11 @@ class TestIntraExtensionAuthzManagerAuthz(tests.TestCase):
self.assertEqual(True, result)
def test_subjects(self):
+ demo_user = self.create_user("demo")
admin_user = self.create_user()
tenant = self.create_tenant()
- ref = self.create_intra_extension("policy_rbac_authz")
- ref_admin = self.create_intra_extension("policy_rbac_admin")
+ ref = self.create_intra_extension("policy_authz")
+ ref_admin = self.create_intra_extension("policy_admin")
self.create_mapping(tenant, ref["id"], ref_admin["id"])
subjects = self.manager.get_subject_dict(admin_user["id"], tenant["id"])
@@ -373,10 +374,11 @@ class TestIntraExtensionAuthzManagerAuthz(tests.TestCase):
admin_user["id"], ref["id"], new_subject["id"])
def test_objects(self):
+ demo_user = self.create_user("demo")
admin_user = self.create_user()
tenant = self.create_tenant()
- ref = self.create_intra_extension("policy_rbac_authz")
- ref_admin = self.create_intra_extension("policy_rbac_admin")
+ ref = self.create_intra_extension("policy_authz")
+ ref_admin = self.create_intra_extension("policy_admin")
self.create_mapping(tenant, ref["id"], ref_admin["id"])
objects = self.manager.get_object_dict(admin_user["id"], tenant["id"])
@@ -408,10 +410,11 @@ class TestIntraExtensionAuthzManagerAuthz(tests.TestCase):
admin_user["id"], ref["id"], new_object["name"])
def test_actions(self):
+ demo_user = self.create_user("demo")
admin_user = self.create_user()
tenant = self.create_tenant()
- ref = self.create_intra_extension("policy_rbac_authz")
- ref_admin = self.create_intra_extension("policy_rbac_admin")
+ ref = self.create_intra_extension("policy_authz")
+ ref_admin = self.create_intra_extension("policy_admin")
self.create_mapping(tenant, ref["id"], ref_admin["id"])
actions = self.manager.get_action_dict(admin_user["id"], tenant["id"])
@@ -443,10 +446,11 @@ class TestIntraExtensionAuthzManagerAuthz(tests.TestCase):
admin_user["id"], ref["id"], new_action["id"])
def test_subject_categories(self):
+ demo_user = self.create_user("demo")
admin_user = self.create_user()
tenant = self.create_tenant()
- ref = self.create_intra_extension("policy_rbac_authz")
- ref_admin = self.create_intra_extension("policy_rbac_admin")
+ ref = self.create_intra_extension("policy_authz")
+ ref_admin = self.create_intra_extension("policy_admin")
self.create_mapping(tenant, ref["id"], ref_admin["id"])
subject_categories = self.manager.get_subject_category_dict(admin_user["id"], ref["id"])
@@ -478,10 +482,11 @@ class TestIntraExtensionAuthzManagerAuthz(tests.TestCase):
admin_user["id"], ref["id"], new_subject_category["name"])
def test_object_categories(self):
+ demo_user = self.create_user("demo")
admin_user = self.create_user()
tenant = self.create_tenant()
- ref = self.create_intra_extension("policy_rbac_authz")
- ref_admin = self.create_intra_extension("policy_rbac_admin")
+ ref = self.create_intra_extension("policy_authz")
+ ref_admin = self.create_intra_extension("policy_admin")
self.create_mapping(tenant, ref["id"], ref_admin["id"])
object_categories = self.manager.get_object_category_dict(admin_user["id"], ref["id"])
@@ -513,10 +518,11 @@ class TestIntraExtensionAuthzManagerAuthz(tests.TestCase):
admin_user["id"], ref["id"], new_object_category["name"])
def test_action_categories(self):
+ demo_user = self.create_user("demo")
admin_user = self.create_user()
tenant = self.create_tenant()
- ref = self.create_intra_extension("policy_rbac_authz")
- ref_admin = self.create_intra_extension("policy_rbac_admin")
+ ref = self.create_intra_extension("policy_authz")
+ ref_admin = self.create_intra_extension("policy_admin")
self.create_mapping(tenant, ref["id"], ref_admin["id"])
action_categories = self.manager.get_action_category_dict(admin_user["id"], ref["id"])
@@ -548,10 +554,11 @@ class TestIntraExtensionAuthzManagerAuthz(tests.TestCase):
admin_user["id"], ref["id"], new_action_category["name"])
def test_subject_category_scope(self):
+ demo_user = self.create_user("demo")
admin_user = self.create_user()
tenant = self.create_tenant()
- ref = self.create_intra_extension("policy_rbac_authz")
- ref_admin = self.create_intra_extension("policy_rbac_admin")
+ ref = self.create_intra_extension("policy_authz")
+ ref_admin = self.create_intra_extension("policy_admin")
self.create_mapping(tenant, ref["id"], ref_admin["id"])
subject_categories = self.admin_manager.set_subject_category_dict(
@@ -596,10 +603,11 @@ class TestIntraExtensionAuthzManagerAuthz(tests.TestCase):
admin_user["id"], ref["id"], subject_category, new_subject_category_scope[new_subject_category_scope_uuid])
def test_object_category_scope(self):
+ demo_user = self.create_user("demo")
admin_user = self.create_user()
tenant = self.create_tenant()
- ref = self.create_intra_extension("policy_rbac_authz")
- ref_admin = self.create_intra_extension("policy_rbac_admin")
+ ref = self.create_intra_extension("policy_authz")
+ ref_admin = self.create_intra_extension("policy_admin")
self.create_mapping(tenant, ref["id"], ref_admin["id"])
object_categories = self.admin_manager.set_object_category_dict(
@@ -644,10 +652,11 @@ class TestIntraExtensionAuthzManagerAuthz(tests.TestCase):
admin_user["id"], ref["id"], object_category, new_object_category_scope[new_object_category_scope_uuid])
def test_action_category_scope(self):
+ demo_user = self.create_user("demo")
admin_user = self.create_user()
tenant = self.create_tenant()
- ref = self.create_intra_extension("policy_rbac_authz")
- ref_admin = self.create_intra_extension("policy_rbac_admin")
+ ref = self.create_intra_extension("policy_authz")
+ ref_admin = self.create_intra_extension("policy_admin")
self.create_mapping(tenant, ref["id"], ref_admin["id"])
action_categories = self.admin_manager.set_action_category_dict(
@@ -692,10 +701,11 @@ class TestIntraExtensionAuthzManagerAuthz(tests.TestCase):
admin_user["id"], ref["id"], action_category, new_action_category_scope[new_action_category_scope_uuid])
def test_subject_category_assignment(self):
+ demo_user = self.create_user("demo")
admin_user = self.create_user()
tenant = self.create_tenant()
- ref = self.create_intra_extension("policy_rbac_authz")
- ref_admin = self.create_intra_extension("policy_rbac_admin")
+ ref = self.create_intra_extension("policy_authz")
+ ref_admin = self.create_intra_extension("policy_admin")
self.create_mapping(tenant, ref["id"], ref_admin["id"])
new_subject = self.create_user()
@@ -792,10 +802,11 @@ class TestIntraExtensionAuthzManagerAuthz(tests.TestCase):
new_subject_category_scope_uuid)
def test_object_category_assignment(self):
+ demo_user = self.create_user("demo")
admin_user = self.create_user()
tenant = self.create_tenant()
- ref = self.create_intra_extension("policy_rbac_authz")
- ref_admin = self.create_intra_extension("policy_rbac_admin")
+ ref = self.create_intra_extension("policy_authz")
+ ref_admin = self.create_intra_extension("policy_admin")
self.create_mapping(tenant, ref["id"], ref_admin["id"])
new_object = {"id": uuid.uuid4().hex, "name": "my_object"}
@@ -892,10 +903,11 @@ class TestIntraExtensionAuthzManagerAuthz(tests.TestCase):
new_object_category_scope_uuid)
def test_action_category_assignment(self):
+ demo_user = self.create_user("demo")
admin_user = self.create_user()
tenant = self.create_tenant()
- ref = self.create_intra_extension("policy_rbac_authz")
- ref_admin = self.create_intra_extension("policy_rbac_admin")
+ ref = self.create_intra_extension("policy_authz")
+ ref_admin = self.create_intra_extension("policy_admin")
self.create_mapping(tenant, ref["id"], ref_admin["id"])
new_action = {"id": uuid.uuid4().hex, "name": "my_action"}
@@ -992,10 +1004,11 @@ class TestIntraExtensionAuthzManagerAuthz(tests.TestCase):
new_action_category_scope_uuid)
def test_sub_meta_rules(self):
+ demo_user = self.create_user("demo")
admin_user = self.create_user()
tenant = self.create_tenant()
- ref = self.create_intra_extension("policy_rbac_authz")
- ref_admin = self.create_intra_extension("policy_rbac_admin")
+ ref = self.create_intra_extension("policy_authz")
+ ref_admin = self.create_intra_extension("policy_admin")
self.create_mapping(tenant, ref["id"], ref_admin["id"])
aggregation_algorithms = self.manager.get_aggregation_algorithms(admin_user["id"], ref["id"])
@@ -1067,10 +1080,11 @@ class TestIntraExtensionAuthzManagerAuthz(tests.TestCase):
admin_user["id"], ref["id"], metarule)
def test_sub_rules(self):
+ demo_user = self.create_user("demo")
admin_user = self.create_user()
tenant = self.create_tenant()
- ref = self.create_intra_extension("policy_rbac_authz")
- ref_admin = self.create_intra_extension("policy_rbac_admin")
+ ref = self.create_intra_extension("policy_authz")
+ ref_admin = self.create_intra_extension("policy_admin")
self.create_mapping(tenant, ref["id"], ref_admin["id"])
sub_meta_rules = self.manager.get_sub_meta_rule(admin_user["id"], ref["id"])