"subject_category_scope",
"object_category_scope",
"action_category_scope",
- "sub_rules",
+ "rules",
"subject_assignments",
"object_assignments",
"action_assignments",
- "sub_meta_rule_relations",
- "aggregation_algorithms",
- "sub_meta_rule"
+ "sub_meta_rule_algorithm",
+ "aggregation_algorithm",
+ "sub_meta_rules"
]
}
name: yyy,
description: bbb},
...},
+ subject_scope_id3: {
+ ...}
subject_category_id2: {...},
...
}
...
}
-aggregation_algorithm = aggregation_algorithm_id
+aggregation_algorithm = {
+ aggregation_algorithm_id: {
+ name: xxx,
+ description: yyy
+ }
+ }
sub_meta_rules = {
sub_meta_rule_id_1: {
def __init__(self):
super(ConfigurationConnector, self).__init__()
self.aggregation_algorithm_dict = dict()
- self.aggregation_algorithm_dict[uuid4().hex] = "all_true"
+ self.aggregation_algorithm_dict[uuid4().hex] = {'name': 'all_true', 'description': 'all_true'}
self.sub_meta_rule_algorithm_dict = dict()
- self.sub_meta_rule_algorithm_dict[uuid4().hex] = "inclusion"
- self.sub_meta_rule_algorithm_dict[uuid4().hex] = "comparison"
+ self.sub_meta_rule_algorithm_dict[uuid4().hex] = {'name': 'inclusion', 'description': 'inclusion'}
+ self.sub_meta_rule_algorithm_dict[uuid4().hex] = {'name': 'comparison', 'description': 'comparison'}
def get_policy_template_dict(self):
nodes = glob(os.path.join(CONF.moon.policy_directory, "*"))
class SubjectAssignment(sql.ModelBase, sql.DictBase):
__tablename__ = 'subject_assignments'
- attributes = ['id', 'subject_assignment', 'intra_extension_id', 'subject_id']
+ attributes = ['id', 'subject_assignment', 'intra_extension_id', 'subject_id', 'subject_category_id']
id = sql.Column(sql.String(64), primary_key=True)
subject_assignment = sql.Column(sql.JsonBlob(), nullable=True)
intra_extension_id = sql.Column(sql.ForeignKey("intra_extensions.id"), nullable=False)
subject_id = sql.Column(sql.ForeignKey("subjects.id"), nullable=False)
+ subject_category_id = sql.Column(sql.ForeignKey("subject_categories.id"), nullable=False)
@classmethod
def from_dict(cls, d):
class AggregationAlgorithm(sql.ModelBase, sql.DictBase):
__tablename__ = 'aggregation_algorithm'
- attributes = ['id', 'aggregation_algorithm_id', 'intra_extension_id']
+ attributes = ['id', 'aggregation_algorithm', 'intra_extension_id']
id = sql.Column(sql.String(64), primary_key=True)
- aggregation_algorithm_id = sql.Column(sql.Text(), nullable=True)
+ aggregation_algorithm = sql.Column(sql.JsonBlob(), nullable=True)
intra_extension_id = sql.Column(sql.ForeignKey("intra_extensions.id"), nullable=False)
@classmethod
# Tenant functions
- def get_intra_extension_dict(self):
+ def get_intra_extensions_dict(self):
with sql.transaction() as session:
query = session.query(IntraExtension.id)
intraextensions = query.all()
- return [intraextension[0] for intraextension in intraextensions]
+ return {intraextension.id: IntraExtension.to_dict(intraextension) for intraextension in intraextensions}
- def set_intra_extension(self, intra_id, intra_extension):
- with sql.transaction() as session:
- # intra_extension["admin"] = jsonutils.dumps(intra_extension["admin"])
- # intra_extension["authz"] = jsonutils.dumps(intra_extension["authz"])
- ie_ref = IntraExtension.from_dict(intra_extension)
- session.add(ie_ref)
- return IntraExtension.to_dict(ie_ref)
-
- def get_intra_extension(self, intra_extension_id):
- with sql.transaction() as session:
- query = session.query(IntraExtension)
- query = query.filter_by(id=intra_extension_id)
- ref = query.first()
- if not ref:
- raise exception.NotFound
- return ref.to_dict()
+ # TODO (dthom): load_intra_extension(self):
def del_intra_extension(self, intra_extension_id):
with sql.transaction() as session:
session.flush()
session.delete(ref)
- def get_intra_extension_name(self, intra_extension_id):
- intra_extension = self.get_intra_extension(intra_extension_id)
- return intra_extension["name"]
-
- def set_intra_extension_name(self, intra_extension_id, intra_extension_name):
- raise exception.NotImplemented() # pragma: no cover
-
- def get_intra_extension_model(self, intra_extension_id):
- intra_extension = self.get_intra_extension(intra_extension_id)
- return intra_extension["model"]
-
- def set_intra_extension_model(self, intra_extension_id, intra_extension_model):
- raise exception.NotImplemented() # pragma: no cover
-
- def get_intra_extension_description(self, intra_extension_id):
- intra_extension = self.get_intra_extension(intra_extension_id)
- return intra_extension["description"]
-
- def set_intra_extension_description(self, intra_extension_id, intra_extension_description):
- raise exception.NotImplemented() # pragma: no cover
+ def set_intra_extension_dict(self, intra_extension_id, intra_extension_dict):
+ with sql.transaction() as session:
+ query = session.query(IntraExtension)
+ query = query.filter_by(id=intra_extension_id)
+ ref = query.first()
+ intra_extension_ref = ref.to_dict()
+ intra_extension_ref.update(intra_extension_dict)
+ new_intra_extension = IntraExtension(
+ id=intra_extension_id,
+ intra_extension=intra_extension_ref
+ )
+ for attr in Tenant.attributes:
+ if attr != 'id':
+ setattr(ref, attr, getattr(new_intra_extension, attr))
+ return IntraExtension.to_dict(ref)
# Getter and Setter for subject_category
- def get_subject_category_dict(self, intra_extension_id):
+ def get_subject_categories_dict(self, intra_extension_id):
with sql.transaction() as session:
query = session.query(SubjectCategory)
query = query.filter_by(intra_extension_id=intra_extension_id)
- ref = query.first()
- if not ref:
- raise IntraExtensionUnknown()
- return ref.to_dict()
+ ref_list = query.all()
+ return {_ref.id: _ref.to_dict()['subject_category'] for _ref in ref_list}
- # TODO: review the code
def set_subject_category_dict(self, intra_extension_id, subject_category_id, subject_category_dict):
with sql.transaction() as session:
query = session.query(SubjectCategory)
query = query.filter_by(intra_extension_id=intra_extension_id)
ref = query.first()
- if not ref:
- raise IntraExtensionUnknown()
- old_ref = ref.to_dict()
- new_ref = SubjectCategory.from_dict(
- {
- "id": subject_category_id,
- 'subject_category': subject_category_dict,
- 'intra_extension_id': intra_extension_id
- }
- )
- for attr in SubjectCategory.attributes:
- if attr != 'id':
- setattr(ref, attr, getattr(new_ref, attr))
- return ref.to_dict()
-
- # TODO: review the code
- def add_subject_category(self, intra_extension_id, subject_category_id, subject_category_name):
- with sql.transaction() as session:
- query = session.query(SubjectCategory)
- query = query.filter_by(intra_extension_id=intra_extension_id)
- ref = query.first()
- subject_category_dict = {
- 'name': subject_category_name,
- 'description': None
- }
new_ref = SubjectCategory.from_dict(
{
"id": subject_category_id,
for attr in SubjectCategory.attributes:
if attr != 'id':
setattr(ref, attr, getattr(new_ref, attr))
- return ref.to_dict()
+ return self.get_subject_categories_dict(intra_extension_id)[subject_category_id]
def del_subject_category(self, intra_extension_id, subject_category_id):
with sql.transaction() as session:
query = session.query(SubjectCategory)
- query = query.filter_by(intra_extension_id=intra_extension_id)
+ query = query.filter_by(subject_category_id=subject_category_id)
ref = query.first()
- if not ref:
- raise IntraExtensionUnknown()
- else:
- old_ref = ref.to_dict()
- subject_category_dict = dict(old_ref["subject_category"])
- try:
- subject_category_dict.pop(subject_category_id)
- except KeyError:
- pass
- else:
- new_ref = SubjectCategory.from_dict(
- {
- "id": old_ref["id"],
- 'subject_categories': subject_category_dict,
- 'intra_extension_id': old_ref["intra_extension_id"]
- }
- )
- for attr in SubjectCategory.attributes:
- if attr != 'id':
- setattr(ref, attr, getattr(new_ref, attr))
- return ref.to_dict()
+ session.delete(ref)
# Getter and Setter for object_category
# Perimeter
- def get_subject_dict(self, intra_extension_id):
+ def get_subjects_dict(self, intra_extension_id):
with sql.transaction() as session:
query = session.query(Subject)
query = query.filter_by(intra_extension_id=intra_extension_id)
- ref = query.first()
- if not ref:
- raise IntraExtensionUnknown()
- return ref.to_dict()
+ ref_list = query.all()
+ return {_ref.id: _ref.to_dict()['subjects'] for _ref in ref_list}
- # TODO: to recheck
- def set_subject_dict(self, intra_extension_id, subject_id):
+ def set_subject_dict(self, intra_extension_id, subject_id, subject_dict):
with sql.transaction() as session:
query = session.query(Subject)
- query = query.filter_by(intra_extension_id=intra_extension_id)
+ query = query.filter_by(subject_id=subject_id)
ref = query.first()
new_ref = Subject.from_dict(
{
- "id": uuid4().hex,
- 'subjects': subject_id,
+ "id": subject_id,
+ 'subjects': subject_dict,
'intra_extension_id': intra_extension_id
}
)
for attr in Subject.attributes:
if attr != 'id':
setattr(ref, attr, getattr(new_ref, attr))
- return ref.to_dict()
-
- def add_subject(self, intra_extension_id, subject_id, subject_name):
- with sql.transaction() as session:
- query = session.query(Subject)
- query = query.filter_by(intra_extension_id=intra_extension_id)
- ref = query.first()
- if not ref:
- raise IntraExtensionUnknown()
- old_ref = ref.to_dict()
- subjects = dict(old_ref["subjects"])
- # TODO: subjects[subject_id] is a dict
- subjects[subject_id] = subject_name
- new_ref = Subject.from_dict(
- {
- "id": old_ref["id"],
- 'subjects': subjects,
- 'intra_extension_id': old_ref["intra_extension_id"]
- }
- )
- for attr in Subject.attributes:
- if attr != 'id':
- setattr(ref, attr, getattr(new_ref, attr))
- return ref.to_dict()
+ return self.get_subjects_dict(intra_extension_id)[subject_id]
def del_subject(self, intra_extension_id, subject_id):
with sql.transaction() as session:
query = session.query(Subject)
- query = query.filter_by(intra_extension_id=intra_extension_id)
+ query = query.filter_by(subject_id=subject_id)
ref = query.first()
- if not ref:
- raise IntraExtensionUnknown()
- else:
- old_ref = ref.to_dict()
- subjects = dict(old_ref["subjects"])
- try:
- subjects.pop(subject_id)
- except KeyError:
- LOG.error("KeyError in remove_subject {} | {}".format(subject_id, subjects))
- else:
- new_ref = Subject.from_dict(
- {
- "id": old_ref["id"],
- 'subjects': subjects,
- 'intra_extension_id': old_ref["intra_extension_id"]
- }
- )
- for attr in Subject.attributes:
- if attr != 'id':
- setattr(ref, attr, getattr(new_ref, attr))
+ session.delete(ref)
def get_object_dict(self, intra_extension_id):
with sql.transaction() as session:
# Getter and Setter for subject_scope
- def get_subject_scope_dict(self, intra_extension_id, subject_category_id):
+ def get_subject_scopes_dict(self, intra_extension_id, subject_category_id):
with sql.transaction() as session:
query = session.query(SubjectScope)
query = query.filter_by(
intra_extension_id=intra_extension_id,
subject_category_id=subject_category_id)
- ref = query.first()
- if not ref:
- raise IntraExtensionUnknown()
- result = copy.deepcopy(ref.to_dict())
- if subject_category_id not in result["subject_scopes"].keys():
- raise SubjectScopeUnknown()
- return result
+ ref_list = query.all()
+ return {_ref.id: _ref.to_dict()['subject_scope'] for _ref in ref_list}
- def set_subject_scope_dict(self, intra_extension_id, subject_category_id, subject_scope_id):
+ def set_subject_scope_dict(self, intra_extension_id, subject_category_id, subject_scope_id, subject_scope_dict):
with sql.transaction() as session:
query = session.query(SubjectScope)
- query = query.filter_by(intra_extension_id=intra_extension_id)
+ query = query.filter_by(subject_scope_id=subject_scope_id)
ref = query.first()
+ new_ref = SubjectScope.from_dict(
+ {
+ "id": subject_scope_id,
+ 'subject_scope': subject_scope_dict,
+ 'intra_extension_id': intra_extension_id,
+ 'subject_category_id': subject_category_id
+ }
+ )
if not ref:
- new_ref = SubjectScope.from_dict(
- {
- "id": uuid4().hex,
- 'subject_scope': {subject_category_id: subject_scope_id},
- 'intra_extension_id': intra_extension_id
- }
- )
session.add(new_ref)
+ ref = new_ref
else:
- tmp_ref = ref.to_dict()
- tmp_ref['subject_scope'].update({subject_category_id: subject_scope_id})
- session.delete(ref)
- new_ref = SubjectScope.from_dict(tmp_ref)
- session.add(new_ref)
- return new_ref.to_dict()
-
- def add_subject_scope(self, intra_extension_id, subject_category_id, subject_scope_id, subject_scope_name):
- with sql.transaction() as session:
- query = session.query(SubjectScope)
- query = query.filter_by(intra_extension_id=intra_extension_id)
- ref = query.first()
- if not ref:
- raise IntraExtensionUnknown()
- old_ref = ref.to_dict()
- scope = copy.deepcopy(old_ref["subject_scope"])
- if subject_category_id not in scope.keys():
- scope[subject_category_id] = dict()
- scope[subject_category_id][subject_scope_id] = subject_scope_name
- return self.set_subject_scope_dict(intra_extension_id, subject_category_id, scope[subject_category_id])
+ for attr in Subject.attributes:
+ if attr != 'id':
+ setattr(ref, attr, getattr(new_ref, attr))
+ return self.get_subject_scopes_dict(intra_extension_id, subject_category_id)[subject_scope_id]
def del_subject_scope(self, intra_extension_id, subject_category_id, subject_scope_id):
with sql.transaction() as session:
query = session.query(SubjectScope)
- query = query.filter_by(intra_extension_id=intra_extension_id)
+ query = query.filter_by(subject_scope_id=subject_scope_id)
ref = query.first()
- if not ref:
- raise IntraExtensionUnknown()
- old_ref = ref.to_dict()
- scope = dict(old_ref["subject_scope"])
- if subject_category_id not in scope:
- return
- try:
- scope[subject_category_id].pop(subject_scope_id)
- except KeyError:
- return
- new_ref = SubjectScope.from_dict(
- {
- "id": old_ref["id"],
- 'subject_scope': scope,
- 'intra_extension_id': old_ref["intra_extension_id"]
- }
- )
- for attr in SubjectScope.attributes:
- if attr != 'id':
- setattr(ref, attr, getattr(new_ref, attr))
- return ref.to_dict()
+ session.delete(ref)
# Getter and Setter for object_category_scope
# Getter and Setter for subject_category_assignment
- def get_subject_assignment_dict(self, intra_extension_id, subject_id):
- """ From a subject_uuid, return a dictionary of (category: scope for that subject)
-
- :param intra_extension_id: intra extension UUID
- :param subject_id: subject UUID
- :return: a dictionary of (keys are category nd values are scope for that subject)
- """
+ def get_subject_assignment_list(self, intra_extension_id, subject_id, subject_category_id):
with sql.transaction() as session:
query = session.query(SubjectAssignment)
- query = query.filter_by(intra_extension_id=intra_extension_id)
+ query = query.filter_by(intra_extension_id=intra_extension_id, subject_id=subject_id, subject_category_id=subject_category_id)
ref = query.first()
- if not ref:
- raise IntraExtensionUnknown()
- _ref = ref.to_dict()
- if subject_id in _ref["subject_assignment"]:
- _backup_dict = _ref["subject_assignment"][subject_id]
- _ref["subject_assignment"] = dict()
- _ref["subject_assignment"][subject_id] = _backup_dict
- else:
- _ref["subject_assignment"] = dict()
- _ref["subject_assignment"][subject_id] = dict()
- return _ref
+ return ref.to_dict()['subject_assignment']
- def set_subject_assignment_dict(self, intra_extension_id, subject_id=None, subject_assignment_dict={}):
+ def set_subject_assignment_list(self, intra_extension_id, subject_id, subject_category_id, subject_assignment_list=[]):
with sql.transaction() as session:
query = session.query(SubjectAssignment)
- query = query.filter_by(intra_extension_id=intra_extension_id)
+ query = query.filter_by(intra_extension_id=intra_extension_id, subject_id=subject_id, subject_category_id=subject_category_id)
ref = query.first()
- if type(subject_assignment_dict) is not dict:
- raise IntraExtensionError()
- for value in subject_assignment_dict.values():
- if type(value) is not list:
- raise IntraExtensionError(str(value))
- if not subject_id:
- subject_assignments = {}
- else:
- subject_assignments = {subject_id: subject_assignment_dict}
new_ref = SubjectAssignment.from_dict(
{
"id": uuid4().hex,
- 'subject_assignment': subject_assignments,
- 'intra_extension_id': intra_extension_id
+ 'subject_assignment': subject_assignment_list,
+ 'intra_extension_id': intra_extension_id,
+ 'subject_id': subject_id,
+ 'subject_category_id': subject_category_id
}
)
if not ref:
session.add(new_ref)
ref = new_ref
else:
- new_ref.subject_assignments[subject_id] = subject_assignment_dict
for attr in SubjectAssignment.attributes:
if attr != 'id':
setattr(ref, attr, getattr(new_ref, attr))
- return ref.to_dict()
+ return self.get_subject_assignment_list(intra_extension_id, subject_id, subject_category_id)
- def add_subject_assignment(self, intra_extension_id, subject_id, subject_category_id, subject_scope_id):
- with sql.transaction() as session:
- query = session.query(SubjectAssignment)
- query = query.filter_by(intra_extension_id=intra_extension_id)
- ref = query.first()
- if not ref:
- raise IntraExtensionUnknown()
- assignment = ref.to_dict()['subject_assignment']
- if subject_id not in assignment:
- assignment[subject_id] = dict()
- if subject_category_id not in assignment[subject_id]:
- assignment[subject_id][subject_category_id] = list()
- if subject_scope_id not in assignment[subject_id][subject_category_id]:
- assignment[subject_id][subject_category_id].append(subject_scope_id)
- return self.set_subject_assignment_dict(
- intra_extension_id,
- subject_id,
- assignment[subject_id])
+ def add_subject_assignment_list(self, intra_extension_id, subject_id, subject_category_id, subject_scope_id):
+ new_subject_assignment_list = self.get_subject_assignment_list(intra_extension_id, subject_id, subject_category_id)
+ new_subject_assignment_list.append(subject_scope_id)
+ return self.set_subject_assignment_list(intra_extension_id, subject_id, subject_category_id, new_subject_assignment_list)
def del_subject_assignment(self, intra_extension_id, subject_id, subject_category_id, subject_scope_id):
- with sql.transaction() as session:
- query = session.query(SubjectAssignment)
- query = query.filter_by(intra_extension_id=intra_extension_id)
- ref = query.first()
- if not ref:
- raise IntraExtensionUnknown()
- old_ref = ref.to_dict()
- if subject_id in old_ref["subject_assignment"]:
- if subject_category_id in old_ref["subject_assignment"][subject_id]:
- old_ref["subject_assignment"][subject_id][subject_category_id].remove(subject_scope_id)
- if not old_ref["subject_assignment"][subject_id][subject_category_id]:
- old_ref["subject_assignment"][subject_id].pop(subject_category_id)
- if not old_ref["subject_assignment"][subject_id]:
- old_ref["subject_assignment"].pop(subject_id)
- try:
- self.set_subject_assignment_dict(
- intra_extension_id,
- subject_id,
- old_ref["subject_assignment"][subject_id])
- except KeyError:
- self.set_subject_assignment_dict(
- intra_extension_id,
- subject_id,
- {})
+ new_subject_assignment_list = self.get_subject_assignment_list(intra_extension_id, subject_id, subject_category_id)
+ new_subject_assignment_list.pop(subject_scope_id)
+ return self.set_subject_assignment_list(intra_extension_id, subject_id, subject_category_id, new_subject_assignment_list)
# Getter and Setter for object_category_assignment
action_id,
old_ref["action_assignment"][action_id])
- # Getter and Setter for meta_rule
+ # Getter and Setter for sub_meta_rule
- def get_sub_meta_rule_dict(self, intra_extension_id):
+ def set_aggregation_algorithm(self, intra_extension_id, aggregation_algorithm_id, aggregation_algorithm_dict):
with sql.transaction() as session:
- query = session.query(SubMetaRule)
+ query = session.query(AggregationAlgorithm)
query = query.filter_by(intra_extension_id=intra_extension_id)
ref = query.first()
+ new_ref = AggregationAlgorithm.from_dict(
+ {
+ "id": aggregation_algorithm_id,
+ 'aggregation_algorithm': aggregation_algorithm_dict,
+ 'intra_extension_id': intra_extension_id
+ }
+ )
if not ref:
- raise IntraExtensionUnknown()
- return ref.to_dict()
+ session.add(new_ref)
+ else:
+ for attr in AggregationAlgorithm.attributes:
+ if attr != 'id':
+ setattr(ref, attr, getattr(new_ref, attr))
+ return self.get_aggregation_algorithm(intra_extension_id)
+
+ def get_aggregation_algorithm(self, intra_extension_id):
+ with sql.transaction() as session:
+ query = session.query(AggregationAlgorithm)
+ query = query.filter_by(intra_extension_id=intra_extension_id)
+ ref = query.first()
+ return {ref.to_dict()['id']: ref.to_dict()['aggregation_algorithm']}
+
+ # Getter and Setter for sub_meta_rule
- def set_sub_meta_rule_dict(self, intra_extension_id, sub_meta_rule_dict):
+ def get_sub_meta_rules_dict(self, intra_extension_id):
+ with sql.transaction() as session:
+ query = session.query(SubMetaRule)
+ query = query.filter_by(intra_extension_id=intra_extension_id)
+ ref_list = query.all()
+ return {_ref.id: _ref.to_dict()['sub_meta_rule'] for _ref in ref_list}
+
+ def set_sub_meta_rule_dict(self, intra_extension_id, sub_meta_rule_id, sub_meta_rule_dict):
with sql.transaction() as session:
query = session.query(SubMetaRule)
query = query.filter_by(intra_extension_id=intra_extension_id)
ref = query.first()
- sub_meta_rule_dict["id"] = uuid4().hex
- sub_meta_rule_dict["intra_extension_id"] = intra_extension_id
- new_ref = SubMetaRule.from_dict(sub_meta_rule_dict)
+ new_ref = SubMetaRule.from_dict(
+ {
+ "id": sub_meta_rule_id,
+ 'sub_meta_rule': sub_meta_rule_dict,
+ 'intra_extension_id': intra_extension_id
+ }
+ )
if not ref:
session.add(new_ref)
- ref = new_ref
else:
for attr in SubMetaRule.attributes:
if attr != 'id':
setattr(ref, attr, getattr(new_ref, attr))
- return ref.to_dict()
+ return self.get_sub_meta_rules_dict(intra_extension_id)
+
+ def del_sub_meta_rule(self, intra_extension_id, sub_meta_rule_id):
+ with sql.transaction() as session:
+ query = session.query(SubMetaRule)
+ query = query.filter_by(sub_meta_rule_id=sub_meta_rule_id)
+ ref = query.first()
+ session.delete(ref)
# Getter and Setter for rules
- def get_rule_dict(self, intra_extension_id):
+ def get_rules_dict(self, intra_extension_id, sub_meta_rule_id):
with sql.transaction() as session:
query = session.query(Rule)
- query = query.filter_by(intra_extension_id=intra_extension_id)
- ref = query.first()
- if not ref:
- raise IntraExtensionUnknown()
- return ref.to_dict()
+ query = query.filter_by(intra_extension_id=intra_extension_id, sub_meta_rule_id=sub_meta_rule_id)
+ ref_list = query.all()
+ return {_ref.id: _ref.to_dict()['rule'] for _ref in ref_list}
- def set_rule_dict(self, intra_extension_id, rule_dict):
+ def set_rule_dict(self, intra_extension_id, sub_meta_rule_id, rule_id, rule_list):
with sql.transaction() as session:
query = session.query(Rule)
- query = query.filter_by(intra_extension_id=intra_extension_id)
+ query = query.filter_by(rule_id=rule_id)
ref = query.first()
- rules = dict()
- rules["id"] = uuid4().hex
- rules["intra_extension_id"] = intra_extension_id
- rules["rule"] = rule_dict
- new_ref = Rule.from_dict(rules)
+ new_ref = Rule.from_dict(
+ {
+ "id": rule_id,
+ 'rule': rule_list,
+ 'intra_extension_id': intra_extension_id,
+ 'sub_meta_rule_id': sub_meta_rule_id
+ }
+ )
if not ref:
session.add(new_ref)
- ref = new_ref
else:
for attr in Rule.attributes:
if attr != 'id':
setattr(ref, attr, getattr(new_ref, attr))
- return ref.to_dict()
+ return self.get_rules_dict(intra_extension_id, sub_meta_rule_id)[rule_id]
+
+ def del_rule_dict(self, intra_extension_id, sub_meta_rule_id, rule_id):
+ with sql.transaction() as session:
+ query = session.query(Rule)
+ query = query.filter_by(rule_id=rule_id)
+ ref = query.first()
+ session.delete(ref)
# class InterExtension(sql.ModelBase, sql.DictBase):
user_id = self._get_user_id_from_token(context.get("token_id"))
# TODO: get tenant name from keystone
tenant_dict = dict()
- tenant_dict['name'] = kw.get("name")
- tenant_dict['description'] = kw.get("description")
- tenant_dict['intra_authz_ext_id'] = kw.get("intra_authz_ext_id")
- tenant_dict['intra_admin_ext_id'] = kw.get("intra_admin_ext_id")
+ tenant_dict['name'] = kw.get("name", None)
+ tenant_dict['description'] = kw.get("description", None)
+ tenant_dict['intra_authz_ext_id'] = kw.get("intra_authz_ext_id", None)
+ tenant_dict['intra_admin_ext_id'] = kw.get("intra_admin_ext_id", None)
return self.tenant_api.add_tenant_dict(user_id, tenant_dict)
@controller.protected()
def get_tenant(self, context, **kw):
user_id = self._get_user_id_from_token(context.get('token_id'))
- tenant_id = kw.get("tenant_id")
+ tenant_id = kw.get("tenant_id", None)
return self.tenant_api.get_tenants_dict(user_id, tenant_id)
@controller.protected()
def del_tenant(self, context, **kw):
user_id = self._get_user_id_from_token(context.get('token_id'))
- tenant_id = kw.get("tenant_id")
+ tenant_id = kw.get("tenant_id", None)
return self.tenant_api.del_tenant(user_id, tenant_id)
def set_tenant(self, context, **kw):
user_id = self._get_user_id_from_token(context.get('token_id'))
- tenant_id = kw["id"]
+ tenant_id = kw.get('id', None)
tenant_dict = dict()
- tenant_dict['name'] = kw.get("name")
- tenant_dict['description'] = kw.get("description")
- tenant_dict['intra_authz_ext_id'] = kw.get("intra_authz_ext_id")
- tenant_dict['intra_admin_ext_id'] = kw.get("intra_admin_ext_id")
+ tenant_dict['name'] = kw.get("name", None)
+ tenant_dict['description'] = kw.get("description", None)
+ tenant_dict['intra_authz_ext_id'] = kw.get("intra_authz_ext_id", None)
+ tenant_dict['intra_admin_ext_id'] = kw.get("intra_admin_ext_id", None)
self.tenant_api.set_tenant_dict(user_id, tenant_id, tenant_dict)
return self.admin_api.load_intra_extension_dict(user_id, intra_extension_dict)
@controller.protected()
- def get_intra_extension_dict(self, context, **kw):
+ def get_intra_extension(self, context, **kw):
user_id = self._get_user_id_from_token(context.get('token_id'))
- ie_id = kw.get('intra_extension_id', None)
- return self.admin_api.get_intra_extensions_dict(user_id)[ie_id]
+ ie_id = kw.get('id', None)
+ return self.admin_api.get_intra_extensions_dict(user_id, ie_id)
@controller.protected()
def del_intra_extension(self, context, **kw):
user_id = self._get_user_id_from_token(context.get('token_id'))
- if "ie_id" not in kw:
- raise IntraExtensionUnknown
- ie_id = kw.get('intra_extension_id', None)
+ ie_id = kw.get('id', None)
self.admin_api.del_intra_extension(user_id, ie_id)
@controller.protected()
def set_intra_extension(self, context, **kw):
user_id = self._get_user_id_from_token(context.get('token_id'))
- if "ie_id" not in kw:
- raise IntraExtensionUnknown
- ie_id = kw.get('intra_extension_id', None)
- self.admin_api.set_intra_extension(user_id, ie_id)
+ ie_id = kw.get('id', None)
+ intra_extension_dict = dict()
+ intra_extension_dict["name"] = kw.get("name", None)
+ intra_extension_dict["model"] = kw.get("model", None)
+ intra_extension_dict["description"] = kw.get("description", None)
+ return self.admin_api.set_intra_extension_dict(user_id, ie_id, intra_extension_dict)
# Metadata functions
@controller.protected()
def get_subject_categories(self, context, **kw):
user_id = self._get_user_id_from_token(context.get('token_id'))
ie_id = kw.get('intra_extension_id', None)
- return self.admin_api.get_subject_category_dict(user_id, ie_id)
+ return self.admin_api.get_subject_categories_dict(user_id, ie_id)
@controller.protected()
def add_subject_category(self, context, **kw):
user_id = self._get_user_id_from_token(context.get('token_id'))
ie_id = kw.get('intra_extension_id', None)
- subject_category_name = kw.get("subject_category_name", None)
- return self.admin_api.add_subject_category(user_id, ie_id, subject_category_name)
+ subject_category_dict = dict()
+ subject_category_dict['name'] = kw.get("name", None)
+ subject_category_dict['description'] = kw.get("description", None)
+ return self.admin_api.add_subject_category(user_id, ie_id, subject_category_dict)
@controller.protected()
def get_subject_category(self, context, **kw):
user_id = self._get_user_id_from_token(context.get('token_id'))
ie_id = kw.get('intra_extension_id', None)
subject_category_id = kw.get("subject_category_id", None)
- return self.admin_api.get_subject_category_dict(user_id, ie_id)[subject_category_id]
+ return self.admin_api.get_subject_categories_dict(user_id, ie_id)[subject_category_id]
@controller.protected()
def del_subject_category(self, context, **kw):
user_id = self._get_user_id_from_token(context.get('token_id'))
ie_id = kw.get('intra_extension_id', None)
- subject_category_id = kw["subject_category_id"]
- return self.admin_api.del_subject_category(user_id, ie_id, subject_category_id)
+ subject_category_id = kw.get("subject_category_id", None)
+ self.admin_api.del_subject_category(user_id, ie_id, subject_category_id)
+
+ @controller.protected()
+ def set_subject_category(self, context, **kw):
+ user_id = self._get_user_id_from_token(context.get('token_id'))
+ ie_id = kw.get('intra_extension_id', None)
+ subject_category_id = kw.get('subject_category_id', None)
+ subject_category_dict = dict()
+ subject_category_dict['name'] = kw.get("name", None)
+ subject_category_dict['description'] = kw.get("description", None)
+ return self.admin_api.set_subject_category(user_id, ie_id, subject_category_id, subject_category_dict)
@controller.protected()
def get_object_categories(self, context, **kw):
def get_subjects(self, context, **kw):
user_id = self._get_user_id_from_token(context.get('token_id'))
ie_id = kw.get('intra_extension_id', None)
- return self.admin_api.get_subject_dict(user_id, ie_id)
+ return self.admin_api.get_subjects_dict(user_id, ie_id)
@controller.protected()
def add_subject(self, context, **kw):
user_id = self._get_user_id_from_token(context.get('token_id'))
ie_id = kw.get('intra_extension_id', None)
subject_name = kw["subject_name"]
- return self.admin_api.add_subject(user_id, ie_id, subject_name)
+ return self.admin_api.add_subject_dict(user_id, ie_id, subject_name)
@controller.protected()
def get_subject(self, context, **kw):
user_id = self._get_user_id_from_token(context.get('token_id'))
ie_id = kw.get('intra_extension_id', None)
subject_id = kw["subject_id"]
- return self.admin_api.get_subject_dict(user_id, ie_id)[subject_id]
+ return self.admin_api.get_subjects_dict(user_id, ie_id)[subject_id]
@controller.protected()
def del_subject(self, context, **kw):
user_id = self._get_user_id_from_token(context.get('token_id'))
ie_id = kw.get('intra_extension_id', None)
subject_id = kw["subject_id"]
- return self.admin_api.del_subject(user_id, ie_id, subject_id)
+ self.admin_api.del_subject(user_id, ie_id, subject_id)
@controller.protected()
def get_objects(self, context, **kw):
def get_subject_scopes(self, context, **kw):
user_id = self._get_user_id_from_token(context.get('token_id'))
ie_id = kw.get("intra_extension_id", None)
- subject_category_id = kw["subject_category_id"]
- return self.admin_api.get_subject_scope_dict(user_id, ie_id, subject_category_id)
+ subject_category_id = kw.get("subject_category_id", None)
+ return self.admin_api.get_subject_scopes_dict(user_id, ie_id, subject_category_id)
@controller.protected()
def add_subject_scope(self, context, **kw):
user_id = self._get_user_id_from_token(context.get('token_id'))
ie_id = kw.get("intra_extension_id", None)
- subject_category_id = kw["subject_category_id"]
- subject_scope_name = kw["subject_scope_name"]
- return self.admin_api.add_subject_scope(
- user_id,
- ie_id,
- subject_category_id,
- subject_scope_name)
+ subject_category_id = kw.get("subject_category_id", None)
+ subject_scope_dict = dict()
+ subject_scope_dict['name'] = kw.get("subject_scope_name", None)
+ subject_scope_dict['description'] = kw.get("subject_scope_description", None)
+ return self.admin_api.add_subject_scope_dict(user_id, ie_id, subject_category_id, subject_scope_dict)
@controller.protected()
def get_subject_scope(self, context, **kw):
user_id = self._get_user_id_from_token(context.get('token_id'))
ie_id = kw.get("intra_extension_id", None)
- subject_category_id = kw["subject_category_id"]
- subject_scope_id = kw["subject_scope_id"]
- return self.admin_api.get_subject_scope(user_id, ie_id, subject_category_id, subject_scope_id)
+ subject_category_id = kw.get("subject_category_id", None)
+ subject_scope_id = kw.get("subject_scope_id", None)
+ return self.admin_api.get_subject_scopes_dict(user_id, ie_id, subject_category_id, subject_scope_id)
@controller.protected()
def del_subject_scope(self, context, **kw):
user_id = self._get_user_id_from_token(context.get('token_id'))
ie_id = kw.get("intra_extension_id", None)
- subject_category_id = kw["subject_category_id"]
- subject_scope_id = kw["subject_scope_id"]
- return self.admin_api.del_subject_scope(
- user_id,
- ie_id,
- subject_category_id,
- subject_scope_id)
+ subject_category_id = kw.get("subject_category_id", None)
+ subject_scope_id = kw.get("subject_scope_id", None)
+ self.admin_api.del_subject_scope(user_id, ie_id, subject_category_id, subject_scope_id)
+
+ @controller.protected()
+ def set_subject_scope(self, context, **kw):
+ user_id = self._get_user_id_from_token(context.get('token_id'))
+ ie_id = kw.get("intra_extension_id", None)
+ subject_category_id = kw.get("subject_category_id", None)
+ subject_scope_id = kw.get("subject_scope_id", None)
+ subject_scope_dict = dict()
+ subject_scope_dict['name'] = kw.get("subject_scope_name", None)
+ subject_scope_dict['description'] = kw.get("subject_scope_description", None)
+ return self.admin_api.set_subject_scope_dict(user_id, ie_id, subject_category_id, subject_scope_id, subject_scope_dict)
@controller.protected()
def get_object_scopes(self, context, **kw):
action_scope_id)
# Assignment functions
- @controller.protected()
- def get_subject_assignments(self, context, **kw):
- user_id = self._get_user_id_from_token(context.get('token_id'))
- ie_id = kw.get("intra_extension_id", None)
- subject_id = kw["subject_id"]
- return self.admin_api.get_subject_assignment_dict(user_id, ie_id, subject_id)
@controller.protected()
def add_subject_assignment(self, context, **kw):
user_id = self._get_user_id_from_token(context.get('token_id'))
ie_id = kw.get("intra_extension_id", None)
- subject_id = kw["subject_id"]
- subject_category_id = kw["subject_category_id"]
- subject_scope_id = kw["subject_scope_id"]
- return self.admin_api.add_subject_assignment(
+ subject_id = kw.get("subject_id", None)
+ subject_category_id = kw.get("subject_category_id", None)
+ subject_scope_id = kw.get("subject_scope_id", None)
+ return self.admin_api.add_subject_assignment_list(
user_id,
ie_id,
subject_id,
def get_subject_assignment(self, context, **kw):
user_id = self._get_user_id_from_token(context.get('token_id'))
ie_id = kw.get("intra_extension_id", None)
- subject_id = kw["subject_id"]
- subject_category_id = kw["subject_category_id"]
- return self.admin_api.get_subject_assignment_dict(user_id, ie_id, subject_id, subject_category_id)
+ subject_id = kw.get("subject_id", None)
+ subject_category_id = kw.get("subject_category_id", None)
+ return self.admin_api.get_subject_assignment_list(user_id, ie_id, subject_id, subject_category_id)
@controller.protected()
def del_subject_assignment(self, context, **kw):
user_id = self._get_user_id_from_token(context.get('token_id'))
ie_id = kw.get("intra_extension_id", None)
- subject_id = kw["subject_id"]
- subject_category_id = kw["subject_category_id"]
- subject_scope_id = kw["subject_scope_id"]
+ subject_id = kw.get("subject_id", None)
+ subject_category_id = kw.get("subject_category_id", None)
+ subject_scope_id = kw.get("subject_scope_id", None)
return self.admin_api.del_subject_assignment(
user_id,
ie_id,
# Metarule functions
@controller.protected()
- def add_aggregation_algorithm(self, context, **kw):
+ def set_aggregation_algorithm(self, context, **kw):
user_id = self._get_user_id_from_token(context.get('token_id'))
ie_id = kw.get("intra_extension_id", None)
- aggregation_algorithm_id = kw["aggregation_algorithm_id"]
- return self.admin_api.add_aggregation_algorithm(
- user_id,
- ie_id,
- aggregation_algorithm_id)
+ aggregation_algorithm_id = kw.get("aggregation_algorithm_id", None)
+ aggregation_algorithm_dict = kw.get("aggregation_algorithm_dict", None)
+ return self.admin_api.set_aggregation_algorithm_dict(user_id, ie_id, aggregation_algorithm_id, aggregation_algorithm_dict)
@controller.protected()
def get_aggregation_algorithm(self, context, **kw):
user_id = self._get_user_id_from_token(context.get('token_id'))
ie_id = kw.get("intra_extension_id", None)
- return self.admin_api.get_aggregation_algorithm(user_id, ie_id)
-
- @controller.protected()
- def del_aggregation_algorithm(self, context, **kw):
- user_id = self._get_user_id_from_token(context.get('token_id'))
- ie_id = kw.get("intra_extension_id", None)
- aggregation_algorithm_id = kw["aggregation_algorithm_id"]
- return self.admin_api.del_aggregation_algorithm(
- user_id,
- ie_id,
- aggregation_algorithm_id)
+ return self.admin_api.get_aggregation_algorithm_dict(user_id, ie_id)
@controller.protected()
def get_sub_meta_rules(self, context, **kw):
user_id = self._get_user_id_from_token(context.get('token_id'))
ie_id = kw.get("intra_extension_id", None)
- return self.admin_api.get_sub_meta_rule_dict(user_id, ie_id)
+ return self.admin_api.get_sub_meta_rules_dict(user_id, ie_id)
@controller.protected()
def add_sub_meta_rule(self, context, **kw):
user_id = self._get_user_id_from_token(context.get('token_id'))
ie_id = kw.get("intra_extension_id", None)
- sub_meta_rule_name = kw["sub_meta_rule_name"]
- subject_category_list = kw["subject_categories"]
- object_category_list = kw["object_categories"]
- action_category_list = kw["action_categories"]
- sub_meta_rule_algorithm = kw["sub_meta_rule_algorithm"]
-
- return self.admin_api.add_sub_meta_rule(
- user_id,
- ie_id,
- sub_meta_rule_name,
- subject_category_list,
- object_category_list,
- action_category_list,
- sub_meta_rule_algorithm
- )
+ sub_meta_rule_dict = dict()
+ sub_meta_rule_dict['name'] = kw.get('sub_meta_rule_name', None)
+ sub_meta_rule_dict['algorithm'] = kw.get('sub_meta_rule_algorithm', None)
+ sub_meta_rule_dict['subject_categories'] = kw.get('sub_meta_rule_subject_categories', None)
+ sub_meta_rule_dict['object_categories'] = kw.get('sub_meta_rule_object_categories', None)
+ sub_meta_rule_dict['action_categories'] = kw.get('sub_meta_rule_action_categories', None)
+ return self.admin_api.add_sub_meta_rule_dict(user_id, ie_id, sub_meta_rule_dict)
@controller.protected()
def get_sub_meta_rule(self, context, **kw):
user_id = self._get_user_id_from_token(context.get('token_id'))
ie_id = kw.get("intra_extension_id", None)
- sub_meta_rule_id = kw["sub_meta_rule_id"]
- return self.admin_api.get_sub_meta_rule(user_id, ie_id, sub_meta_rule_id)
+ sub_meta_rule_id = kw.get("sub_meta_rule_id", None)
+ return self.admin_api.get_sub_meta_rules_dict(user_id, ie_id, sub_meta_rule_id)
@controller.protected()
def del_sub_meta_rule(self, context, **kw):
user_id = self._get_user_id_from_token(context.get('token_id'))
ie_id = kw.get("intra_extension_id")
- sub_meta_rule_id = kw["sub_meta_rule_id"]
- return self.admin_api.get_sub_meta_rule(user_id, ie_id, sub_meta_rule_id)
+ sub_meta_rule_id = kw.get("sub_meta_rule_id", None)
+ self.admin_api.del_sub_meta_rule(user_id, ie_id, sub_meta_rule_id)
+
+ @controller.protected()
+ def set_sub_meta_rule(self, context, **kw):
+ user_id = self._get_user_id_from_token(context.get('token_id'))
+ ie_id = kw.get("intra_extension_id", None)
+ sub_meta_rule_id = kw.get("sub_meta_rule_id", None)
+ sub_meta_rule_dict = dict()
+ sub_meta_rule_dict['name'] = kw.get('sub_meta_rule_name', None)
+ sub_meta_rule_dict['algorithm'] = kw.get('sub_meta_rule_algorithm', None)
+ sub_meta_rule_dict['subject_categories'] = kw.get('sub_meta_rule_subject_categories', None)
+ sub_meta_rule_dict['object_categories'] = kw.get('sub_meta_rule_object_categories', None)
+ sub_meta_rule_dict['action_categories'] = kw.get('sub_meta_rule_action_categories', None)
+ return self.admin_api.set_sub_meta_rule_dict(user_id, ie_id, sub_meta_rule_id, sub_meta_rule_dict)
# Rules functions
@controller.protected()
def get_rules(self, context, **kw):
user_id = self._get_user_id_from_token(context.get('token_id'))
ie_id = kw.get("intra_extension_id", None)
- sub_meta_rule_id = kw["sub_meta_rule_id"]
- return self.admin_api.get_rule_dict(user_id, ie_id, sub_meta_rule_id)
+ sub_meta_rule_id = kw.get("sub_meta_rule_id", None)
+ return self.admin_api.get_rules_dict(user_id, ie_id, sub_meta_rule_id)
@controller.protected()
def add_rule(self, context, **kw):
user_id = self._get_user_id_from_token(context.get('token_id'))
ie_id = kw.get("intra_extension_id", None)
- sub_meta_rule_id = kw.get("sub_meta_rule_id")
+ sub_meta_rule_id = kw.get("sub_meta_rule_id", None)
rule_list = list()
- subject_category_list = kw.get('subject_categories')
- object_category_list = kw.get('object_categories')
- action_category_list = kw.get('action_categories')
- rule_list = subject_category_list + object_category_list + action_category_list
+ subject_category_list = kw.get('subject_categories', [])
+ object_category_list = kw.get('object_categories', [])
+ action_category_list = kw.get('action_categories', [])
+ rule_list = subject_category_list + action_category_list + object_category_list
return self.admin_api.add_rule_list(user_id, ie_id, sub_meta_rule_id, rule_list)
@controller.protected()
def get_rule(self, context, **kw):
user_id = self._get_user_id_from_token(context.get('token_id'))
ie_id = kw.get("intra_extension_id", None)
- sub_meta_rule_id = kw.get("sub_meta_rule_id")
- rule_id = kw.get("rule_id")
- return self.admin_api.get_rule(user_id, ie_id, sub_meta_rule_id, rule_id)
+ sub_meta_rule_id = kw.get("sub_meta_rule_id", None)
+ rule_id = kw.get("rule_id", None)
+ return self.admin_api.get_rules_dict(user_id, ie_id, sub_meta_rule_id, rule_id)
@controller.protected()
def del_rule(self, context, **kw):
user_id = self._get_user_id_from_token(context.get('token_id'))
ie_id = kw.get("intra_extension_id", None)
- sub_meta_rule_id = kw["sub_meta_rule_id"]
- rule_id = kw["rule_id"]
- return self.admin_api.del_rule(user_id, ie_id, sub_meta_rule_id, rule_id)
+ sub_meta_rule_id = kw.get("sub_meta_rule_id", None)
+ rule_id = kw.get("rule_id", None)
+ self.admin_api.del_rule(user_id, ie_id, sub_meta_rule_id, rule_id)
+
+ @controller.protected()
+ def set_rule(self, context, **kw):
+ user_id = self._get_user_id_from_token(context.get('token_id'))
+ ie_id = kw.get("intra_extension_id", None)
+ sub_meta_rule_id = kw.get("sub_meta_rule_id", None)
+ rule_id = kw.get("rule_id", None)
+ rule_list = list()
+ subject_category_list = kw.get('subject_categories', [])
+ object_category_list = kw.get('object_categories', [])
+ action_category_list = kw.get('action_categories', [])
+ rule_list = subject_category_list + action_category_list + object_category_list
+ return self.admin_api.set_rule_list(user_id, ie_id, sub_meta_rule_id, rule_id, rule_list)
@dependency.requires('authz_api')
return wrapped
-def enforce(actions, object, **extra):
- _actions = actions
+def enforce(action_names, object_name, **extra):
+ _action_name_list = action_names
def wrap(func):
def wrapped(*args):
# global actions
self = args[0]
user_name = args[1]
- intra_extension_uuid = args[2]
- _admin_extension_uuid = self.tenant_api.get_admin_extension_uuid(args[2])
+ intra_extension_id = args[2]
+ if intra_extension_id not in self.admin_api.get_intra_extensions(DEFAULT_USER):
+ raise IntraExtensionUnknown()
+
+ tenants_dict = self.tenant_api.get_tenants_dict(DEFAULT_USER)
+ intra_admin_extension_id = None
+ tenant_name = None
+ for tenant_id in tenants_dict:
+ if tenants_dict[tenant_id]['intra_authz_extension_id'] is intra_extension_id:
+ intra_admin_extension_id = tenants_dict[tenant_id]['intra_admin_extension_id']
+ tenant_name = tenants_dict[tenant_id]['name']
+
# func.func_globals["_admin_extension_uuid"] = _admin_extension_uuid
- if not _admin_extension_uuid:
+ if not intra_admin_extension_id:
args[0].moonlog_api.warning("No admin IntraExtension found, authorization granted by default.")
return func(*args)
else:
- _authz = False
- if type(_actions) in (str, unicode):
- actions = (_actions, )
+ authz_result = False
+ if type(_action_name_list) in (str, unicode):
+ action_name_list = (_action_name_list, )
else:
- actions = _actions
- for action in actions:
- if self.admin_api.authz(
- _admin_extension_uuid,
- user_name,
- object,
- action):
- _authz = True
+ action_name_list = _action_name_list
+ for action_name in action_name_list:
+ if self.authz_api.authz(tenant_name, user_name, object_name, action_name, 'admin'):
+ authz_result = True
else:
- _authz = False
+ authz_result = False
break
- if _authz:
+ if authz_result:
return func(*args)
return wrapped
return wrap
raise TenantNoIntraExtension()
return tenant_dict[tenant_id][genre]
- # TODO: check if we need the func
- def get_tenant_uuid(self, extension_uuid):
- for _tenant_uuid, _tenant_value in six.iteritems(self.get_tenants_dict()):
- if extension_uuid == _tenant_value["authz"] or extension_uuid == _tenant_value["admin"]:
- return _tenant_uuid
- raise TenantIDNotFound()
-
- # TODO: check if we need the func
- def get_admin_extension_uuid(self, authz_extension_uuid):
- _tenants = self.get_tenants_dict()
- for _tenant_uuid in _tenants:
- 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)))
- # FIXME (dthom): if AdminIntraExtensionNotFound, maybe we can add an option in configuration file
- # to allow or not the fact that Admin IntraExtension can be None
- # raise AdminIntraExtensionNotFound()
-
- # TODO: check if we need the func
- def delete(self, authz_extension_uuid):
- _tenants = self.get_tenants_dict()
- for _tenant_uuid in _tenants:
- if authz_extension_uuid == _tenants[_tenant_uuid]['authz']:
- return self.set_tenant_dict(_tenant_uuid, "", "", "")
- raise AuthzIntraExtensionNotFound(_("No IntraExtension found mapping this Authz IntraExtension: {}.".format(
- authz_extension_uuid)))
-
@dependency.requires('identity_api', 'tenant_api', 'configuration_api', 'authz_api', 'admin_api', 'moonlog_api')
class IntraExtensionManager(manager.Manager):
- __genre__ = None
-
def __init__(self):
+ self.__genre__ = None
driver = CONF.moon.intraextension_driver
super(IntraExtensionManager, self).__init__(driver)
authz_buffer['object_id'] = object_id
authz_buffer['action_id'] = action_id
meta_data_dict = dict()
- meta_data_dict["subject_categories"] = self.driver.get_subject_category_dict(intra_extension_id)
+ meta_data_dict["subject_categories"] = self.driver.get_subject_categories_dict(intra_extension_id)
meta_data_dict["object_categories"] = self.driver.get_object_category_dict(intra_extension_id)
meta_data_dict["action_categories"] = self.driver.get_action_category_dict(intra_extension_id)
subject_assignment_dict = dict()
for category in meta_data_dict["subject_categories"]:
- subject_assignment_dict[category] = self.driver.get_subject_assignment_dict(
+ subject_assignment_dict[category] = self.driver.get_subject_assignment_list(
intra_extension_id, subject_id)[category]
object_assignment_dict = dict()
for category in meta_data_dict["object_categories"]:
authz_buffer = self.__get_authz_buffer(intra_extension_id, subject_id, object_id, action_id)
decision_buffer = dict()
- meta_rule_dict = self.driver.get_sub_meta_rule_dict(intra_extension_id)
+ meta_rule_dict = self.driver.get_sub_meta_rules_dict(intra_extension_id)
for sub_meta_rule_id in meta_rule_dict['sub_meta_rules']:
if meta_rule_dict['sub_meta_rules'][sub_meta_rule_id]['algorithm'] == 'inclusion':
decision_buffer[sub_meta_rule_id] = inclusion(
authz_buffer,
meta_rule_dict['sub_meta_rules'][sub_meta_rule_id],
- self.driver.get_rule_dict(intra_extension_id, sub_meta_rule_id).values())
+ self.driver.get_rules_dict(intra_extension_id, sub_meta_rule_id).values())
elif meta_rule_dict['sub_meta_rules'][sub_meta_rule_id]['algorithm'] == 'comparison':
decision_buffer[sub_meta_rule_id] = comparison(
authz_buffer,
meta_rule_dict['sub_meta_rules'][sub_meta_rule_id],
- self.driver.get_rule_dict(intra_extension_id, sub_meta_rule_id).values())
+ self.driver.get_rules_dict(intra_extension_id, sub_meta_rule_id).values())
if meta_rule_dict['aggregation'] == 'all_true':
return all_true(decision_buffer)
)
# Note (dthom): subject_category_assignment must be initialized because when there is no data in json
# we will not go through the for loop
- self.driver.set_subject_assignment_dict(intra_extension_dict["id"])
+ self.driver.set_subject_assignment_list(intra_extension_dict["id"])
for subject in subject_assignments:
- self.driver.set_subject_assignment_dict(intra_extension_dict["id"], subject, subject_assignments[subject])
+ self.driver.set_subject_assignment_list(intra_extension_dict["id"], subject, subject_assignments[subject])
object_assignments = dict()
for category_name, value in json_assignments["object_assignments"].iteritems():
intra_extension_dict["rule"] = {"rule": copy.deepcopy(json_rules)}
# Translate value from JSON file to UUID for Database
rules = dict()
- sub_meta_rules = self.driver.get_sub_meta_rule_dict(intra_extension_dict["id"])
+ sub_meta_rules = self.driver.get_sub_meta_rules_dict(intra_extension_dict["id"])
for relation in json_rules:
# print(relation)
# print(self.get_sub_meta_rule_relations("admin", ie["id"]))
ie_dict["name"] = filter_input(intra_extension_dict["name"])
ie_dict["model"] = filter_input(intra_extension_dict["policymodel"])
ie_dict["description"] = filter_input(intra_extension_dict["description"])
- ref = self.driver.set_intra_extension(ie_dict['id'], ie_dict)
+ ref = self.driver.set_intra_extension_dict(ie_dict['id'], ie_dict)
self.moonlog_api.debug("Creation of IE: {}".format(ref))
# read the profile given by "policymodel" and populate default variables
policy_dir = os.path.join(CONF.moon.policy_directory, ie_dict["model"])
self.__load_rule_file(ie_dict, policy_dir)
return ref
- def get_intra_extension(self, user_id, intra_extension_id):
+ def get_intra_extension_dict(self, user_id, intra_extension_id):
"""
:param user_id:
:return: {intra_extension_id: intra_extension_name, ...}
raise IntraExtensionUnknown()
return self.driver.del_intra_extension(intra_extension_id)
+ def set_intra_extension_dict(self, user_id, intra_extension_id, intra_extension_dict):
+ # TODO: check will be done through super_extension later
+ if intra_extension_id not in self.driver.get_intra_extensions_dict():
+ raise IntraExtensionUnknown()
+ return self.driver.set_intra_extension_dict(intra_extension_id, intra_extension_dict)
+
# Metadata functions
@filter_args # TODO: check for each function if intra_entension_id exists
@enforce("read", "subject_categories")
- def get_subject_category_dict(self, user_id, intra_extension_id):
+ def get_subject_categories_dict(self, user_id, intra_extension_id):
"""
:param user_id:
:param intra_extension_id:
subject_category_id2: {...},
...}
"""
- return self.driver.get_subject_category_dict(intra_extension_id)
+ return self.driver.get_subject_categories_dict(intra_extension_id)
@filter_args
@enforce(("read", "write"), "subject_categories")
- @enforce(("read", "write"), "subject_scopes")
- def add_subject_category(self, user_id, intra_extension_id, subject_category_name):
- subject_category_dict = self.driver.get_subject_category_dict(intra_extension_id)
+ def add_subject_category(self, user_id, intra_extension_id, subject_category_dict):
+ subject_category_dict = self.driver.get_subject_categories_dict(intra_extension_id)
for subject_category_id in subject_category_dict:
- if subject_category_dict[subject_category_id]['name'] is subject_category_name:
+ if subject_category_dict[subject_category_id]['name'] is subject_category_dict['name']:
raise SubjectCategoryNameExisting()
- subject_category_id = uuid4().hex
- # TODO (dthom): create category in scope
- # self.driver.create_subject_category_in_scope(intra_extension_id, subject_category_id)
- return self.driver.add_subject_category(intra_extension_id, subject_category_id, subject_category_name)
+ return self.driver.set_subject_category(intra_extension_id, uuid4().hex, subject_category_dict)
@filter_args
@enforce("read", "subject_categories")
def get_subject_category(self, user_id, intra_extension_id, subject_category_id):
- subject_category_dict = self.driver.get_subject_category_dict(intra_extension_id)
+ subject_category_dict = self.driver.get_subject_categories_dict(intra_extension_id)
if subject_category_id not in subject_category_dict:
raise SubjectCategoryUnknown()
return subject_category_dict[subject_category_id]
@enforce(("read", "write"), "subject_scopes")
@enforce(("read", "write"), "subject_assignments")
def del_subject_category(self, user_id, intra_extension_id, subject_category_id):
- if subject_category_id not in self.driver.get_subject_category_dict(intra_extension_id):
+ if subject_category_id not in self.driver.get_subject_categories_dict(intra_extension_id):
raise SubjectCategoryUnknown()
# TODO (dthom): destroy category in scope
# self.driver.destroy_subject_category_in_scope(intra_extension_id, subject_category_id)
# self.driver.destroy_subject_category_in_assignement(intra_extension_id, subject_category_id)
return self.driver.del_subject_category(intra_extension_id, subject_category_id)
+ @filter_args
+ @enforce(("read", "write"), "subject_categories")
+ def set_subject_category(self, user_id, intra_extension_id, subject_category_id, subject_category_dict):
+ for subject_category_id not in self.driver.get_subject_categories_dict(intra_extension_id):
+ raise SubjectCategoryUnknown()
+ return self.driver.set_subject_category(intra_extension_id, subject_category_id, subject_category_dict)
+
@filter_args
@enforce("read", "object_categories")
def get_object_category_dict(self, user_id, intra_extension_id):
@filter_args
@enforce("read", "subjects")
- def get_subject_dict(self, user_id, intra_extension_id):
- """
- :param user_id:
- :param intra_extension_id:
- :return: {
- subject_id1: {
- name: xxx,
- description: yyy,
- ...},
- subject_id2: {...},
- ...}
- """
- return self.driver.get_subject_dict(intra_extension_id)
+ def get_subjects_dict(self, user_id, intra_extension_id):
+ return self.driver.get_subjects_dict(intra_extension_id)
@filter_args
@enforce(("read", "write"), "subjects")
- def add_subject(self, user_id, intra_extension_id, subject_name):
- subject_dict = self.driver.get_subject_dict(intra_extension_id)
- for subject_id in subject_dict:
- if subject_dict[subject_id]["name"] is subject_name:
+ def add_subject_dict(self, user_id, intra_extension_id, subject_dict):
+ subjects_dict = self.driver.get_subjects_dict(intra_extension_id)
+ for subject_id in subjects_dict:
+ if subjects_dict[subject_id]["name"] is subject_dict['name']:
raise SubjectNameExisting()
# Next line will raise an error if user is not present in Keystone database
- subject_item_dict = self.identity_api.get_user_by_name(subject_name, "default")
- return self.driver.add_subject(intra_extension_id, subject_item_dict["id"], subject_name)
+ subject_item_dict = self.identity_api.get_user_by_name(subject_dict['name'], "default")
+ return self.driver.set_subject_dict(intra_extension_id, subject_item_dict["id"], subject_dict)
@filter_args
@enforce("read", "subjects")
- def get_subject(self, user_id, intra_extension_id, subject_id):
- subject_dict = self.driver.get_subject_dict(intra_extension_id)
+ def get_subject_dict(self, user_id, intra_extension_id, subject_id):
+ subject_dict = self.driver.get_subjects_dict(intra_extension_id)
if subject_id in subject_dict:
raise SubjectUnknown()
return subject_dict[subject_id]
@filter_args
@enforce(("read", "write"), "subjects")
def del_subject(self, user_id, intra_extension_id, subject_id):
- if subject_id in self.driver.get_subject_dict(intra_extension_id):
+ if subject_id in self.driver.get_subjects_dict(intra_extension_id):
raise SubjectUnknown()
# TODO (dthom): destroy item-related assignment
- return self.driver.del_subject(intra_extension_id, subject_id)
+ self.driver.del_subject(intra_extension_id, subject_id)
+
+ @filter_args
+ @enforce(("read", "write"), "subjects")
+ def set_subject_dict(self, user_id, intra_extension_id, subject_id, subject_dict):
+ subjects_dict = self.driver.get_subjects_dict(intra_extension_id)
+ for subject_id in subjects_dict:
+ if subjects_dict[subject_id]["name"] is subject_dict['name']:
+ raise SubjectNameExisting()
+ # Next line will raise an error if user is not present in Keystone database
+ subject_item_dict = self.identity_api.get_user_by_name(subject_dict['name'], "default")
+ return self.driver.set_subject_dict(intra_extension_id, subject_item_dict["id"], subject_dict)
@filter_args
@enforce("read", "objects")
@filter_args
@enforce("read", "subject_scopes")
@enforce("read", "subject_categories")
- def get_subject_scope_dict(self, user_id, intra_extension_id, subject_category_id):
+ def get_subject_scopes_dict(self, user_id, intra_extension_id, subject_category_id):
"""
:param user_id:
:param intra_extension_id:
des: bbb},
...}
"""
- if subject_category_id not in self.driver.get_subject_category_dict(intra_extension_id):
+ if subject_category_id not in self.driver.get_subject_categories_dict(intra_extension_id):
raise SubjectCategoryUnknown()
- return self.driver.get_subject_scope_dict(intra_extension_id, subject_category_id)
+ return self.driver.get_subject_scopes_dict(intra_extension_id, subject_category_id)
@filter_args
@enforce(("read", "write"), "subject_scopes")
@enforce("read", "subject_categories")
- def add_subject_scope(self, user_id, intra_extension_id, subject_category_id, subject_scope_name):
- if subject_category_id not in self.driver.get_subject_category_dict(intra_extension_id):
+ def add_subject_scope_dict(self, user_id, intra_extension_id, subject_category_id, subject_scope_dict):
+ if subject_category_id not in self.driver.get_subject_categories_dict(intra_extension_id):
raise SubjectCategoryUnknown()
- subject_scope_dict = self.driver.get_subject_scope_dict(intra_extension_id, subject_category_id)
- for _subject_scope_id in subject_scope_dict:
- if subject_scope_name is subject_scope_dict[_subject_scope_id]['name']:
+ subject_scopes_dict = self.driver.get_subject_scopes_dict(intra_extension_id, subject_category_id)
+ for _subject_scope_id in subject_scopes_dict:
+ if subject_scope_dict['name'] is subject_scopes_dict[_subject_scope_id]['name']:
raise SubjectScopeNameExisting()
subject_scope_id = uuid4().hex
- return self.driver.add_subject_scope(
- intra_extension_id,
- subject_category_id,
- subject_scope_id,
- subject_scope_name)
+ return self.driver.add_subject_scope_dict(intra_extension_id, subject_category_id, subject_scope_id, subject_scope_dict)
@filter_args
@enforce("read", "subject_scopes")
@enforce("read", "subject_categories")
- def get_subject_scope(self, user_id, intra_extension_id, subject_category_id, subject_scope_id):
- if subject_category_id not in self.driver.get_subject_category_dict(intra_extension_id):
+ def get_subject_scope_dict(self, user_id, intra_extension_id, subject_category_id, subject_scope_id):
+ if subject_category_id not in self.driver.get_subject_categories_dict(intra_extension_id):
raise SubjectCategoryUnknown()
- subject_scopte_dict = self.driver.get_subject_scope_dict(intra_extension_id, subject_category_id)
- if subject_scope_id not in subject_scopte_dict:
+ subject_scopes_dict = self.driver.get_subject_scopes_dict(intra_extension_id, subject_category_id)
+ if subject_scope_id not in subject_scopes_dict:
raise SubjectScopeUnknown()
- return subject_scopte_dict[subject_scope_id]
+ return subject_scopes_dict[subject_scope_id]
@filter_args
@enforce(("read", "write"), "subject_scopes")
@enforce("read", "subject_categories")
def del_subject_scope(self, user_id, intra_extension_id, subject_category_id, subject_scope_id):
- if subject_category_id not in self.driver.get_subject_category_dict(intra_extension_id):
+ if subject_category_id not in self.driver.get_subject_categories_dict(intra_extension_id):
raise SubjectCategoryUnknown()
- if subject_scope_id not in self.driver.get_subject_scope_dict(intra_extension_id, subject_category_id):
+ if subject_scope_id not in self.driver.get_subject_scopes_dict(intra_extension_id, subject_category_id):
raise SubjectScopeUnknown()
# TODO (dthom): destroy scope-related assignment
# TODO (dthom): destroy scope-related rule
return self.driver.del_subject_scope(intra_extension_id, subject_category_id, subject_scope_id)
+ @filter_args
+ @enforce(("read", "write"), "subject_scopes")
+ @enforce("read", "subject_categories")
+ def set_subject_scope_dict(self, user_id, intra_extension_id, subject_category_id, subject_scope_id, subject_scope_name):
+ if subject_category_id not in self.driver.get_subject_categories_dict(intra_extension_id):
+ raise SubjectCategoryUnknown()
+ subject_scope_dict = self.driver.get_subject_scopes_dict(intra_extension_id, subject_category_id)
+ for _subject_scope_id in subject_scope_dict:
+ if subject_scope_name is subject_scope_dict[_subject_scope_id]['name']:
+ raise SubjectScopeNameExisting()
+ return self.driver.set_subject_scope_dict(intra_extension_id, subject_category_id, uuid4().hex, subject_scope_dict)
+
@filter_args
@enforce("read", "object_category_scopes")
@enforce("read", "object_categories")
if object_scope_name is object_scope_dict[_object_scope_id]['name']:
raise ObjectScopeNameExisting()
object_scope_id = uuid4().hex
- return self.driver.add_subject_scope(
+ return self.driver.add_subject_scope_dict(
intra_extension_id,
object_category_id,
object_scope_id,
# Assignment functions
- @filter_args
- @enforce("read", "subject_assignments")
- @enforce("read", "subjects")
- def get_subject_assignment_dict(self, user_id, intra_extension_id, subject_id):
- """
- :param user_id:
- :param intra_extension_id:
- :param subject_id:
- :return: {
- subject_category_id1: [subject_scope_id1, subject_scope_id2, ...],
- subject_category_id2: [subject_scope_id1, subject_scope_id2, ...],
- ...
- }
- """
- if subject_id not in self.driver.get_subject_dict(user_id, intra_extension_id):
- raise SubjectUnknown()
- return self.driver.get_subject_assignment_dict(intra_extension_id, subject_id)
-
@filter_args
@enforce(("read", "write"), "subject_assignments")
@enforce("read", "subjects")
@enforce("read", "subject_categories")
@enforce("read", "subject_scopes")
def add_subject_assignment(self, user_id, intra_extension_id, subject_id, subject_category_id, subject_scope_id):
- if subject_id not in self.driver.get_subject_dict(intra_extension_id):
+ if subject_id not in self.driver.get_subjects_dict(intra_extension_id):
raise SubjectUnknown()
- elif subject_category_id not in self.driver.get_subject_category_dict(intra_extension_id):
+ if subject_category_id not in self.driver.get_subject_categories_dict(intra_extension_id):
raise SubjectCategoryUnknown()
- elif subject_scope_id not in self.driver.get_subject_scope_dict(intra_extension_id, subject_category_id):
+ if subject_scope_id not in self.driver.get_subject_scopes_dict(intra_extension_id, subject_category_id):
raise SubjectScopeUnknown()
- elif subject_scope_id in self.driver.get_subject_assignment_dict(intra_extension_id, subject_id)[subject_category_id]:
+ if subject_scope_id in self.driver.get_subject_assignment_list(intra_extension_id, subject_id, subject_category_id):
raise SubjectAssignmentExisting()
- return self.driver.add_subject_assignment(intra_extension_id, subject_id, subject_category_id, subject_scope_id)
+ return self.driver.add_subject_assignment_list(intra_extension_id, subject_id, subject_category_id, subject_scope_id)
@filter_args
@enforce("read", "subject_assignments")
@enforce("read", "subjects")
@enforce("read", "subject_categories")
- def get_subject_assignment(self, user_id, intra_extension_id, subject_id, subject_category_id):
- if subject_id not in self.driver.get_subject_dict(user_id, intra_extension_id):
+ def get_subject_assignment_list(self, user_id, intra_extension_id, subject_id, subject_category_id):
+ if subject_id not in self.driver.get_subjects_dict(user_id, intra_extension_id):
raise SubjectUnknown()
- elif subject_category_id not in self.driver.get_subject_category_dict(intra_extension_id):
+ elif subject_category_id not in self.driver.get_subject_categories_dict(intra_extension_id):
raise SubjectCategoryUnknown()
- return self.driver.get_subject_assignment_dict(intra_extension_id, subject_id)[subject_category_id]
+ return self.driver.get_subject_assignment_list(intra_extension_id, subject_id)[subject_category_id]
@filter_args
@enforce(("read", "write"), "subject_assignments")
@enforce("read", "subject_categories")
@enforce("read", "subject_scopes")
def del_subject_assignment(self, user_id, intra_extension_id, subject_id, subject_category_id, subject_scope_id):
- if subject_id not in self.driver.get_subject_dict(intra_extension_id):
+ if subject_id not in self.driver.get_subjects_dict(intra_extension_id):
raise SubjectUnknown()
- elif subject_category_id not in self.driver.get_subject_category_dict(intra_extension_id):
+ elif subject_category_id not in self.driver.get_subject_categories_dict(intra_extension_id):
raise SubjectCategoryUnknown()
- elif subject_scope_id not in self.driver.get_subject_scope_dict(intra_extension_id, subject_category_id):
+ elif subject_scope_id not in self.driver.get_subject_scopes_dict(intra_extension_id, subject_category_id):
raise SubjectScopeUnknown()
- elif subject_scope_id not in self.driver.get_subject_assignment_dict(intra_extension_id, subject_id)[subject_category_id]:
+ elif subject_scope_id not in self.driver.get_subject_assignment_list(intra_extension_id, subject_id)[subject_category_id]:
raise SubjectAssignmentUnknown()
- return self.driver.del_subject_category_assignment(intra_extension_id, subject_id, subject_category_id, subject_scope_id)
+ self.driver.del_subject_category_assignment(intra_extension_id, subject_id, subject_category_id, subject_scope_id)
@filter_args
@enforce("read", "object_assignments")
raise ObjectCategoryUnknown()
elif object_scope_id not in self.driver.get_object_scope_dict(intra_extension_id, object_category_id):
raise ObjectScopeUnknown()
- elif object_scope_id not in self.driver.get_subject_assignment_dict(intra_extension_id, object_id)[object_category_id]:
+ elif object_scope_id not in self.driver.get_subject_assignment_list(intra_extension_id, object_id)[object_category_id]:
raise ObjectAssignmentUnknown()
return self.driver.del_object_assignment(intra_extension_id, object_id, object_category_id, object_scope_id)
@filter_args
@enforce(("read", "write"), "aggregation_algorithm")
- def add_aggregation_algorithm(self, user_id, intra_extension_id, aggregation_algorithm_id):
- if aggregation_algorithm_id not in self.configuration_api.get_aggregation_algorithms:
- raise AggregationAlgorithmUnknown()
- elif self.dirver.get_aggregation_algorithm(intra_extension_id):
- raise AggregationAlgorithmExisting()
- return self.driver.add_aggregation_algorithm(intra_extension_id, aggregation_algorithm_id)
+ def set_aggregation_algorithm_dict(self, user_id, intra_extension_id, aggregation_algorithm_id, aggregation_algorithm_dict):
+ if aggregation_algorithm_id:
+ if aggregation_algorithm_id not in self.configuration_api.get_aggregation_algorithms(DEFAULT_USER):
+ raise AggregationAlgorithmUnknown()
+ else:
+ aggregation_algorithm_id = uuid4().hex
+ return self.driver.set_aggregation_algorithm_dict(intra_extension_id, aggregation_algorithm_id, aggregation_algorithm_dict)
@filter_args
@enforce("read", "aggregation_algorithm")
- def get_aggregation_algorithm(self, user_id, intra_extension_id):
- if not self.dirver.get_aggregation_algorithm(intra_extension_id):
- raise AggregationAlgorithmNotExisting()
- return self.driver.get_aggregation_algorithm(intra_extension_id)
-
- @filter_args
- @enforce("write", "aggregation_algorithm")
- def del_aggregation_algorithm(self, user_id, intra_extension_id, aggregation_algorithm_id):
- if aggregation_algorithm_id is not self.dirver.get_aggregation_algorithm(intra_extension_id):
+ def get_aggregation_algorithm_dict(self, user_id, intra_extension_id):
+ """
+ :param user_id:
+ :param intra_extension_id:
+ :return: {
+ aggregation_algorithm_id: {
+ name: xxx,
+ description: yyy
+ }
+ }
+ """
+ if not self.driver.get_aggregation_algorithm_dict(intra_extension_id):
raise AggregationAlgorithmNotExisting()
- return self.driver.del_aggregation_algorithm(intra_extension_id, aggregation_algorithm_id)
+ return self.driver.get_aggregation_algorithm_dict(intra_extension_id)
@filter_args
@enforce("read", "sub_meta_rules")
- def get_sub_meta_rule_dict(self, user_id, intra_extension_id):
+ def get_sub_meta_rules_dict(self, user_id, intra_extension_id):
"""
:param user_id:
:param intra_extension_id:
...
}
"""
- return self.driver.get_sub_meta_rule_dict(intra_extension_id)
+ return self.driver.get_sub_meta_rules_dict(intra_extension_id)
@filter_args
@enforce(("read", "write"), "sub_meta_rules")
@enforce("write", "rule")
- def add_sub_meta_rule(self,
- user_id,
- intra_extension_id,
- sub_meta_rule_name,
- subject_category_list,
- object_category_list,
- action_category_list,
- sub_meta_rule_algorithm):
- sub_meta_rule_dict = self.driver.get_sub_meta_rule_dict(intra_extension_id)
- for _sub_meta_rule_id in sub_meta_rule_dict:
- if sub_meta_rule_name is sub_meta_rule_dict[_sub_meta_rule_id]["name"]:
+ def add_sub_meta_rule_dict(self, user_id, intra_extension_id, sub_meta_rule_dict):
+ sub_meta_rules_dict = self.driver.get_sub_meta_rules_dict(intra_extension_id)
+ for _sub_meta_rule_id in sub_meta_rules_dict:
+ if sub_meta_rule_dict['name'] is sub_meta_rules_dict[_sub_meta_rule_id]["name"]:
raise SubMetaRuleNameExisting()
- elif subject_category_list is sub_meta_rule_dict[_sub_meta_rule_id]["subject_categories"] and \
- object_category_list is sub_meta_rule_dict[_sub_meta_rule_id]["object_categories"] and \
- action_category_list is sub_meta_rule_dict[_sub_meta_rule_id]["action_categories"] and \
- sub_meta_rule_algorithm is sub_meta_rule_dict[_sub_meta_rule_id]["algorithm"]:
+ elif sub_meta_rule_dict['subject_categories'] is sub_meta_rules_dict[_sub_meta_rule_id]["subject_categories"] and \
+ sub_meta_rule_dict['object_categories'] is sub_meta_rules_dict[_sub_meta_rule_id]["object_categories"] and \
+ sub_meta_rule_dict['action_categories'] is sub_meta_rules_dict[_sub_meta_rule_id]["action_categories"] and \
+ sub_meta_rule_dict['algorithm'] is sub_meta_rules_dict[_sub_meta_rule_id]["algorithm"]:
raise SubMetaRuleExisting()
sub_meta_rule_id = uuid4().hex()
# TODO (dthom): add new sub-meta-rule to rule
# self.driver.add_rule(intra_extension_id, sub_meta_rule_id, [])
- return self.driver.add_sub_meta_rule(
- intra_extension_id,
- sub_meta_rule_id,
- sub_meta_rule_name,
- subject_category_list,
- object_category_list,
- action_category_list,
- sub_meta_rule_algorithm)
+ return self.driver.set_sub_meta_rule_dict(intra_extension_id, sub_meta_rule_id, sub_meta_rule_dict)
@filter_args
@enforce(("read", "write"), "sub_meta_rules")
- def get_sub_meta_rule(self, user_id, intra_extension_id, sub_meta_rule_id):
- sub_meta_rule_dict = self.driver.get_sub_meta_rule_dict(intra_extension_id)
+ def get_sub_meta_rule_dict(self, user_id, intra_extension_id, sub_meta_rule_id):
+ sub_meta_rule_dict = self.driver.get_sub_meta_rules_dict(intra_extension_id)
if sub_meta_rule_id not in sub_meta_rule_dict:
raise SubMetaRuleUnknown()
return sub_meta_rule_dict[sub_meta_rule_id]
@enforce(("read", "write"), "sub_meta_rules")
@enforce(("read", "write"), "rule")
def del_sub_meta_rule(self, user_id, intra_extension_id, sub_meta_rule_id):
- if sub_meta_rule_id not in self.driver.get_sub_meta_rule_dict(intra_extension_id):
+ if sub_meta_rule_id not in self.driver.get_sub_meta_rules_dict(intra_extension_id):
raise SubMetaRuleUnknown()
# TODO (dthom): destroy sub-meta-rule-related rules
# self.driver.del_rule(intra_extension_id, sub_meta_rule_id, "*")
- return self.driver.del_sub_meta_rule(intra_extension_id, sub_meta_rule_id)
+ self.driver.del_sub_meta_rule(intra_extension_id, sub_meta_rule_id)
+
+ @filter_args
+ @enforce(("read", "write"), "sub_meta_rules")
+ @enforce("write", "rule")
+ def set_sub_meta_rule_dict(self, user_id, intra_extension_id, sub_meta_rule_id, sub_meta_rule_dict):
+ if sub_meta_rule_id not in self.driver.get_sub_meta_rules_dict(intra_extension_id):
+ raise SubMetaRuleUnknown()
+ # TODO (dthom): add new sub-meta-rule to rule
+ # self.driver.add_rule(intra_extension_id, sub_meta_rule_id, [])
+ return self.driver.set_sub_meta_rule_dict(intra_extension_id, sub_meta_rule_id, sub_meta_rule_dict)
# Rule functions
@filter_args
- @enforce("read", "rule")
- def get_rule_dict(self, user_id, intra_extension_id, sub_meta_rule_id):
+ @enforce("read", "rules")
+ def get_rules_dict(self, user_id, intra_extension_id, sub_meta_rule_id):
"""
:param user_id:
:param intra_extension_id:
rule_id2: [subject_scope3, subject_scope4, ..., action_scope3, ..., object_scope3, ... ],
...}
"""
- return self.driver.get_rule_dict(intra_extension_id, sub_meta_rule_id)
+ return self.driver.get_rules_dict(intra_extension_id, sub_meta_rule_id)
@filter_args
@enforce("read", "sub_meta_rules")
- @enforce(("read", "write"), "rule")
- def add_rule(self, user_id, intra_extension_id, sub_meta_rule_id, rule_list):
- if sub_meta_rule_id not in self.driver.get_sub_meta_rule_dict(intra_extension_id):
+ @enforce(("read", "write"), "rules")
+ def add_rule_dict(self, user_id, intra_extension_id, sub_meta_rule_id, rule_list):
+ if sub_meta_rule_id not in self.driver.get_sub_meta_rules_dict(intra_extension_id):
raise SubMetaRuleUnknown()
- elif rule_list in self.driver.get_rule_dict(intra_extension_id, sub_meta_rule_id).values():
+ if rule_list in self.driver.get_rules_dict(intra_extension_id, sub_meta_rule_id).values():
raise RuleExisting()
- rule_id = uuid4().hex()
- return self.driver.add_rule(intra_extension_id, sub_meta_rule_id, rule_id, rule_list)
+ rule_id = uuid4().hex
+ return self.driver.set_rule_dict(intra_extension_id, sub_meta_rule_id, rule_id, rule_list)
@filter_args
@enforce("read", "sub_meta_rules")
- @enforce("read", "rule")
- def get_rule(self, user_id, intra_extension_id, sub_meta_rule_id, rule_id):
- if sub_meta_rule_id not in self.driver.get_sub_meta_rule_dict(intra_extension_id):
+ @enforce("read", "rules")
+ def get_rule_dict(self, user_id, intra_extension_id, sub_meta_rule_id, rule_id):
+ if sub_meta_rule_id not in self.driver.get_sub_meta_rules_dict(intra_extension_id):
raise SubMetaRuleUnknown()
- rule_dict = self.driver.get_rule_dict(intra_extension_id)
- if rule_id not in rule_dict[sub_meta_rule_id]:
+ rules_dict = self.driver.get_rules_dict(intra_extension_id, sub_meta_rule_id)
+ if rule_id not in rules_dict:
raise RuleUnknown()
- return rule_dict[rule_id]
+ return rules_dict[rule_id]
@filter_args
@enforce("read", "sub_meta_rules")
- @enforce(("read", "write"), "rule")
+ @enforce(("read", "write"), "rules")
def del_rule(self, user_id, intra_extension_id, sub_meta_rule_id, rule_id):
- if sub_meta_rule_id not in self.driver.get_sub_meta_rule_dict(intra_extension_id):
+ if sub_meta_rule_id not in self.driver.get_sub_meta_rules_dict(intra_extension_id):
raise SubMetaRuleUnknown()
- rule_dict = self.driver.get_rule_dict(intra_extension_id, sub_meta_rule_id)
- if rule_id not in rule_dict:
+ rules_dict = self.driver.get_rules_dict(intra_extension_id, sub_meta_rule_id)
+ if rule_id not in rules_dict:
raise RuleUnknown()
- return self.driver.del_rule(intra_extension_id, sub_meta_rule_id, rule_id)
+ self.driver.del_rule(intra_extension_id, sub_meta_rule_id, rule_id)
+ @filter_args
+ @enforce("read", "sub_meta_rules")
+ @enforce(("read", "write"), "rules")
+ def set_rule_dict(self, user_id, intra_extension_id, sub_meta_rule_id, rule_id, rule_list):
+ if sub_meta_rule_id not in self.driver.get_sub_meta_rules_dict(intra_extension_id):
+ raise SubMetaRuleUnknown()
+ rules_dict = self.driver.get_rules_dict(intra_extension_id, sub_meta_rule_id)
+ if rule_id not in rules_dict:
+ raise RuleUnknown()
+ return self.driver.set_rule_dict(intra_extension_id, sub_meta_rule_id, rule_id, rule_list)
@dependency.provider('authz_api')
@dependency.requires('identity_api', 'tenant_api', 'moonlog_api')
class IntraExtensionAuthzManager(IntraExtensionManager):
- __genre__ = "authz"
+ def __init__(self):
+ self.__genre__ = "authz"
+ super(IntraExtensionAuthzManager, self).__init__()
- def authz(self, tenant_name, subject_name, object_name, action_name):
+ def authz(self, tenant_name, subject_name, object_name, action_name, genre="authz"):
# TODO (dthom) add moon log
"""Check authorization for a particular action.
:return: True or False or raise an exception
"""
tenant_id = self.tenant_api.get_tenant_id_from_name(DEFAULT_USER, tenant_name)
- intra_extension_id = self.tenant_api.get_tenant_intra_extension_id(DEFAULT_USER, tenant_id, self.__genre__)
- subjects_dict = self.driver.get_subject_dict(intra_extension_id)
+ intra_extension_id = self.tenant_api.get_tenant_intra_extension_id(DEFAULT_USER, tenant_id, genre)
+ subjects_dict = self.driver.get_subjects_dict(intra_extension_id)
subject_id = None
for _subject_id in subjects_dict:
if subjects_dict[_subject_id]['name'] is subject_name:
raise ActionUnknown()
return super(IntraExtensionAuthzManager, self).authz(intra_extension_id, subject_id, object_id, action_id)
- def del_intra_extension(self, intra_extension_id):
+ def del_intra_extension(self, user_id, intra_extension_id):
raise AdminException()
def set_subject_dict(self, user_id, intra_extension_uuid, subject_dict):
raise SubjectAddNotAuthorized()
- def add_subject(self, user_id, intra_extension_uuid, subject_name):
+ def add_subject_dict(self, user_id, intra_extension_uuid, subject_name):
raise SubjectAddNotAuthorized()
def del_subject(self, user_id, intra_extension_uuid, subject_name):
def set_subject_scope_dict(self, user_id, intra_extension_uuid, category, scope):
raise SubjectCategoryScopeAddNotAuthorized()
- def add_subject_scope(self, user_id, intra_extension_uuid, subject_category, scope_name):
+ def add_subject_scope_dict(self, user_id, intra_extension_uuid, subject_category, scope_name):
raise SubjectCategoryScopeAddNotAuthorized()
def del_subject_scope(self, user_id, intra_extension_uuid, subject_category, subject_category_scope):
def add_action_assignment(self, user_id, intra_extension_uuid, action_uuid, category_uuid, scope_uuid):
raise ActionCategoryAssignmentDelNotAuthorized()
- def set_aggregation_algorithm(self, user_id, intra_extension_uuid, aggregation_algorithm):
+ def set_aggregation_algorithm_dict(self, user_id, intra_extension_uuid, aggregation_algorithm):
raise MetaRuleAddNotAuthorized()
- def get_sub_meta_rule(self, user_id, intra_extension_uuid, sub_meta_rules):
+ def get_sub_meta_rule_dict(self, user_id, intra_extension_uuid, sub_meta_rules):
raise MetaRuleAddNotAuthorized()
def set_sub_rule(self, user_id, intra_extension_uuid, relation, sub_rule):
@dependency.requires('identity_api', 'tenant_api', 'moonlog_api')
class IntraExtensionAdminManager(IntraExtensionManager):
- __genre__ = "admin"
+ def __init__(self):
+ self.__genre__ = "admin"
+ super(IntraExtensionAdminManager, self).__init__()
@dependency.provider('moonlog_api')
data_values = list()
if data_name == self.SUBJECT:
- data_values = self.get_subject_dict(intra_extension_uuid)["subjects"]
+ data_values = self.get_subjects_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()
(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"]
+ data_values = self.get_subject_categories_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()
# Getter and Setter for intra_extension
- def get_intra_extension_dict(self):
- """Get a list of IntraExtension
- :return: {"intra_extension_id": "intra_extension_name", }
- """
- raise exception.NotImplemented() # pragma: no cover
-
- def get_intra_extension(self, intra_extension_id):
- """Get a description of an IntraExtension
- :param intra_extension_id: the IntraExtension UUID
- :return: {intra_extension_id: intra_extension_name, ...}
- """
+ def get_intra_extensions_dict(self):
raise exception.NotImplemented() # pragma: no cover
- def set_intra_extension(self, intra_extension_id, intra_extension_dict):
- """Set a new IntraExtension
-
- :param intra_extension_id: IntraExtension UUID
- :type intra_extension_id: string
- :param intra_extension_dict: a dictionary withe the description of the IntraExtension (see below)
- :type intra_extension_dict: dict
- :return: the IntraExtension dictionary, example:
- {
- "id": "uuid1",
- "name": "Name of the intra_extension",
- "model": "Model of te intra_extension (admin or authz)"
- "description": "a description of the intra_extension"
- }
- """
- raise exception.NotImplemented() # pragma: no cover
+ # TODO: check with load
+ # def add_intra_extensions_dict(self):
+ # raise exception.NotImplemented() # pragma: no cover
def del_intra_extension(self, intra_extension_id):
- """Delete an IntraExtension
+ raise exception.NotImplemented() # pragma: no cover
- :param intra_extension_id: IntraExtension UUID
- :type intra_extension_id: string
- :return: None
- """
+ def set_intra_extension_dict(self, intra_extension_id, intra_extension_dict):
raise exception.NotImplemented() # pragma: no cover
# Metadata functions
- def get_subject_category_dict(self, intra_extension_id):
- """Get a list of all subject categories
-
- :param intra_extension_id: IntraExtension UUID
- :type intra_extension_id: string
- :return: a dictionary containing all subject categories {"uuid1": "name1", "uuid2": "name2"}
- """
+ def get_subject_categories_dict(self, intra_extension_id):
raise exception.NotImplemented() # pragma: no cover
def set_subject_category_dict(self, intra_extension_id, subject_category_id, subject_category_dict):
- """Set the list of all subject categories
-
- :param intra_extension_id: IntraExtension UUID
- :type intra_extension_id: string
- :param subject_category_dict: dict of subject categories {"uuid1": "name1", "uuid2": "name2"}
- :type subject_category_dict: dict
- :return: a dictionary containing all subject categories {"uuid1": "name1", "uuid2": "name2"}
- """
- raise exception.NotImplemented() # pragma: no cover
-
- def add_subject_category(self, intra_extension_id, subject_category_id, subject_category_name):
- """Add a subject category
-
- :param intra_extension_id: IntraExtension UUID
- :type intra_extension_id: string
- :param subject_category_id: the UUID of the subject category
- :type subject_category_id: string
- :param subject_category_name: the name of the subject category
- :type subject_category_name: string
- :return: a dictionnary with the subject catgory added {"uuid1": "name1"}
- """
raise exception.NotImplemented() # pragma: no cover
def del_subject_category(self, intra_extension_id, subject_category_id):
- """Remove one subject category
-
- :param intra_extension_id: IntraExtension UUID
- :type intra_extension_id: string
- :param subject_category_id: the UUID of subject category to remove
- :type subject_category_id: string
- :return: a dictionary containing all subject categories {"uuid1": "name1", "uuid2": "name2"}
- """
raise exception.NotImplemented() # pragma: no cover
def get_object_category_dict(self, intra_extension_id):
# Perimeter functions
- def get_subject_dict(self, intra_extension_id):
+ def get_subjects_dict(self, intra_extension_id):
"""Get the list of subject for that IntraExtension
:param intra_extension_id: IntraExtension UUID
"""
raise exception.NotImplemented() # pragma: no cover
- def set_subject_dict(self, intra_extension_id, subject_dict):
- """Set the list of subject for that IntraExtension
-
- :param intra_extension_id: IntraExtension UUID
- :type intra_extension_id: string
- :param subject_dict: dict of subject: {"uuid1": "name1", "uuid2": "name2"}
- :type subject_dict: dict
- :return: a dictionary containing all subjects for that IntraExtension, eg. {"uuid1": "name1", "uuid2": "name2"}
- """
- raise exception.NotImplemented() # pragma: no cover
-
- def add_subject(self, intra_extension_id, subject_id, subject_name):
- """Add a subject
-
- :param intra_extension_id: IntraExtension UUID
- :type intra_extension_id: string
- :param subject_id: Subject UUID
- :type subject_id: string
- :param subject_name: Subject name
- :type subject_name: string
- :return: the added subject {"uuid1": "name1"}
- """
+ def set_subject_dict(self, intra_extension_id, subject_id, subject_dict):
raise exception.NotImplemented() # pragma: no cover
def del_subject(self, intra_extension_id, subject_id):
- """Remove a subject
-
- :param intra_extension_id: IntraExtension UUID
- :type intra_extension_id: string
- :param subject_id: Subject UUID
- :type subject_id: string
- :return: None
- """
raise exception.NotImplemented() # pragma: no cover
def get_object_dict(self, intra_extension_id):
- """Get the list of object for that IntraExtension
-
- :param intra_extension_id: IntraExtension UUID
- :type intra_extension_id: string
- :return: a dictionary containing all objects for that IntraExtension, eg. {"uuid1": "name1", "uuid2": "name2"}
- """
raise exception.NotImplemented() # pragma: no cover
def set_object_dict(self, intra_extension_id, object_dict):
- """Set the list of object for that IntraExtension
-
- :param intra_extension_id: IntraExtension UUID
- :type intra_extension_id: string
- :param object_dict: dict of object: {"uuid1": "name1", "uuid2": "name2"}
- :type object_dict: dict
- :return: a dictionary containing all objects for that IntraExtension, eg. {"uuid1": "name1", "uuid2": "name2"}
- """
raise exception.NotImplemented() # pragma: no cover
def add_object(self, intra_extension_id, object_id, object_name):
- """Ad an object
-
- :param intra_extension_id: IntraExtension UUID
- :type intra_extension_id: string
- :param object_id: Object UUID
- :type object_id: string
- :param object_name: Object name
- :type object_name: string
- :return: the added object {"uuid1": "name1"}
- """
raise exception.NotImplemented() # pragma: no cover
def del_object(self, intra_extension_id, object_id):
- """Remove an object
-
- :param intra_extension_id: IntraExtension UUID
- :type intra_extension_id: string
- :param object_id: Object UUID
- :type object_id: string
- :return: None
- """
raise exception.NotImplemented() # pragma: no cover
def get_action_dict(self, intra_extension_id):
- """ Get the list of action for that IntraExtension
-
- :param intra_extension_id: IntraExtension UUID
- :type intra_extension_id: string
- :return: a dictionary containing all actions for that IntraExtension, eg. {"uuid1": "name1", "uuid2": "name2"}
- """
raise exception.NotImplemented() # pragma: no cover
def set_action_dict(self, intra_extension_id, action_dict):
- """ Set the list of action for that IntraExtension
-
- :param intra_extension_id: IntraExtension UUID
- :type intra_extension_id: string
- :param action_dict: dict of actions: {"uuid1": "name1", "uuid2": "name2"}
- :type action_dict: dict
- :return: a dictionary containing all actions for that IntraExtension, eg. {"uuid1": "name1", "uuid2": "name2"}
- """
raise exception.NotImplemented() # pragma: no cover
def add_action(self, intra_extension_id, action_id, action_name):
- """Ad an action
-
- :param intra_extension_id: IntraExtension UUID
- :type intra_extension_id: string
- :param action_id: Action UUID
- :type action_id: string
- :param action_name: Action name
- :type action_name: string
- :return: the added action {"uuid1": "name1"}
- """
raise exception.NotImplemented() # pragma: no cover
def del_action(self, intra_extension_id, action_id):
- """Remove an action
-
- :param intra_extension_id: IntraExtension UUID
- :type intra_extension_id: string
- :param action_id: Action UUID
- :type action_id: string
- :return: None
- """
raise exception.NotImplemented() # pragma: no cover
# Scope functions
- def get_subject_scope_dict(self, intra_extension_id, subject_category_id):
- """Get a list of all subject category scope
-
- :param intra_extension_id: IntraExtension UUID
- :type intra_extension_id: string
- :param subject_category_id: the category UUID where the scope values are
- :type subject_category_id: string
- :return: a dictionary containing all subject category scope {"category1": {"scope_uuid1": "scope_name1}}
- """
- raise exception.NotImplemented() # pragma: no cover
-
- def set_subject_scope_dict(self, intra_extension_id, subject_category_id, subject_scope_dict):
- """Set the list of all scope for that subject category
-
- :param intra_extension_id: IntraExtension UUID
- :type intra_extension_id: string
- :param subject_category_id: the UUID of the subject category where this scope will be set
- :type subject_category_id: string
- :return: a dictionary containing all scope {"scope_uuid1": "scope_name1, "scope_uuid2": "scope_name2}
- """
+ def get_subject_scopes_dict(self, intra_extension_id, subject_category_id):
raise exception.NotImplemented() # pragma: no cover
- def add_subject_scope(self, intra_extension_id, subject_category_id, subject_scope_id, subject_scope_name):
- """Add a subject category
-
- :param intra_extension_id: IntraExtension UUID
- :type intra_extension_id: string
- :param subject_category_id: the subject category UUID where the scope will be added
- :type subject_category_id: string
- :param subject_scope_id: the UUID of the subject category
- :type subject_scope_id: string
- :param subject_scope_name: the name of the subject category
- :type subject_scope_name: string
- :return: a dictionary containing the subject category scope added {"category1": {"scope_uuid1": "scope_name1}}
- """
+ def set_subject_scope_dict(self, intra_extension_id, subject_category_id, subject_scope_id, subject_scope_dict):
raise exception.NotImplemented() # pragma: no cover
def del_subject_scope(self, intra_extension_id, subject_category_id, subject_scope_id):
- """Remove one scope belonging to a subject category
-
- :param intra_extension_id: IntraExtension UUID
- :type intra_extension_id: string
- :param subject_category_id: the UUID of subject categorywhere we can find the scope to remove
- :type subject_category_id: string
- :param subject_scope_id: the UUID of the scope to remove
- :type subject_scope_id: string
- :return: None
- """
raise exception.NotImplemented() # pragma: no cover
def get_object_scope_dict(self, intra_extension_id, object_category_id):
# Assignment functions
- def get_subject_assignment_dict(self, intra_extension_id, subject_id):
- """Get the assignment for a given subject_uuid
-
- :param intra_extension_id: IntraExtension UUID
- :type intra_extension_id: string
- :param subject_id: subject UUID
- :type subject_id: string
- :return: a dictionary of assignment for the given subject {"cat1": ["scope_uuid1", "scope_uuid2"]}
- :raises: IntraExtensionNotFound, SubjectUnknown, SubjectCategoryAssignmentUnknown, SubjectCategoryAssignmentOutOfScope
- """
+ def get_subject_assignment_list(self, intra_extension_id, subject_id, subject_category_id):
raise exception.NotImplemented() # pragma: no cover
- def set_subject_assignment_dict(self, intra_extension_id, subject_id, subject_assignment_dict):
- """Set the assignment for a given subject_uuid
-
- :param intra_extension_id: IntraExtension UUID
- :type intra_extension_id: string
- :param subject_id: subject UUID
- :type subject_id: string
- :param subject_assignment_dict: the assignment dictionary {"cat1": ["scope_uuid1", "scope_uuid2"]}
- :type subject_assignment_dict: dict
- :return: a dictionary of assignment for the given subject {"cat1": ["scope_uuid1", "scope_uuid2"]}
- """
+ def set_subject_assignment_list(self, intra_extension_id, subject_id, subject_category_id, subject_assignment_list):
raise exception.NotImplemented() # pragma: no cover
- def add_subject_assignment(self, intra_extension_id, subject_id, subject_category_id, subject_scope_id):
- """Add a scope to a category and to a subject
-
- :param intra_extension_id: IntraExtension UUID
- :type intra_extension_id: string
- :param subject_id: the subject UUID
- :type subject_id: string
- :param subject_category_id: the category UUID
- :type subject_category_id: string
- :param subject_scope_id: the scope UUID
- :type subject_scope_id: string
- :return: a dictionary of assignment for the given subject {"cat1": ["scope_uuid1", "scope_uuid2"]}
- """
+ def add_subject_assignment_list(self, intra_extension_id, subject_id, subject_category_id, subject_scope_id):
raise exception.NotImplemented() # pragma: no cover
def del_subject_assignment(self, intra_extension_id, subject_id, subject_category_id, subject_scope_id):
- """Remove a scope from a category and from a subject
-
- :param intra_extension_id: IntraExtension UUID
- :type intra_extension_id: string
- :param subject_id: the subject UUID
- :type subject_id: string
- :param subject_category_id: the category UUID
- :type subject_category_id: string
- :param subject_scope_id: the scope UUID
- :type subject_scope_id: string
- :return: None
- """
raise exception.NotImplemented() # pragma: no cover
def get_object_assignment_dict(self, intra_extension_id, object_id):
- """Get the assignment for a given object_uuid
-
- :param intra_extension_id: IntraExtension UUID
- :type intra_extension_id: string
- :param object_id: object UUID
- :type object_id: string
- :return: a dictionary of assignment for the given object {"cat1": ["scope_uuid1", "scope_uuid2"]}
- :raises: IntraExtensionNotFound, ObjectUnknown, ObjectCategoryAssignmentUnknown, ObjectCategoryAssignmentOutOfScope
- """
raise exception.NotImplemented() # pragma: no cover
def set_object_assignment_dict(self, intra_extension_id, object_id, object_assignment_dict):
- """Set the assignment for a given object_uuid
-
- :param intra_extension_id: IntraExtension UUID
- :type intra_extension_id: string
- :param object_id: object UUID
- :type object_id: string
- :param object_assignment_dict: the assignment dictionary {"cat1": ["scope_uuid1", "scope_uuid2"]}
- :type object_assignment_dict: dict
- :return: a dictionary of assignment for the given object {"cat1": ["scope_uuid1", "scope_uuid2"]}
- """
raise exception.NotImplemented() # pragma: no cover
def add_object_assignment(self, intra_extension_id, object_id, object_category_id, object_scope_id):
- """Add a scope to a category and to a object
-
- :param intra_extension_id: IntraExtension UUID
- :type intra_extension_id: string
- :param object_id: the object UUID
- :type object_id: string
- :param object_category_id: the category UUID
- :type object_category_id: string
- :param object_scope_id: the scope UUID
- :type object_scope_id: string
- :return: a dictionary of assignment for the given object {"cat1": ["scope_uuid1", "scope_uuid2"]}
- """
raise exception.NotImplemented() # pragma: no cover
def del_object_assignment(self, intra_extension_id, object_id, object_category_id, object_scope_id):
- """Remove a scope from a category and from a object
-
- :param intra_extension_id: IntraExtension UUID
- :type intra_extension_id: string
- :param object_id: the object UUID
- :type object_id: string
- :param object_category_id: the category UUID
- :type object_category_id: string
- :param object_scope_id: the scope UUID
- :type object_scope_id: string
- :return: None
- """
raise exception.NotImplemented() # pragma: no cover
def get_action_assignment_dict(self, intra_extension_id, action_id):
- """Get the assignment for a given action_uuid
-
- :param intra_extension_id: IntraExtension UUID
- :type intra_extension_id: string
- :param action_id: action UUID
- :type action_id: string
- :return: a dictionary of assignment for the given action {"cat1": ["scope_uuid1", "scope_uuid2"]}
- :raises: IntraExtensionNotFound, ActionUnknown, ActionCategoryAssignmentUnknown, ActionCategoryAssignmentOutOfScope
- """
raise exception.NotImplemented() # pragma: no cover
def set_action_assignment_dict(self, intra_extension_id, action_id, action_assignment_dict):
- """Set the assignment for a given action_uuid
-
- :param intra_extension_id: IntraExtension UUID
- :type intra_extension_id: string
- :param action_id: action UUID
- :type action_id: string
- :param action_assignment_dict: the assignment dictionary {"cat1": ["scope_uuid1", "scope_uuid2"]}
- :type action_assignment_dict: dict
- :return: a dictionary of assignment for the given action {"cat1": ["scope_uuid1", "scope_uuid2"]}
- """
raise exception.NotImplemented() # pragma: no cover
def add_action_assignment(self, intra_extension_id, action_id, action_category_id, action_scope_id):
- """Add a scope to a category and to a action
-
- :param intra_extension_id: IntraExtension UUID
- :type intra_extension_id: string
- :param action_id: the action UUID
- :type action_id: string
- :param action_category_id: the category UUID
- :type action_category_id: string
- :param action_scope_id: the scope UUID
- :type action_scope_id: string
- :return: a dictionary of assignment for the given action {"cat1": ["scope_uuid1", "scope_uuid2"]}
- """
raise exception.NotImplemented() # pragma: no cover
def del_action_assignment(self, intra_extension_id, action_id, action_category_id, action_scope_id):
- """Remove a scope from a category and from a action
-
- :param intra_extension_id: IntraExtension UUID
- :type intra_extension_id: string
- :param action_id: the action UUID
- :type action_id: string
- :param action_category_id: the category UUID
- :type action_category_id: string
- :param action_scope_id: the scope UUID
- :type action_scope_id: string
- :return: None
- """
raise exception.NotImplemented() # pragma: no cover
# Meta_rule functions
- def get_sub_meta_rule_dict(self, extension_uuid):
- """Get the Meta rule
-
- :param extension_uuid: IntraExtension UUID
- :type extension_uuid: string
- :return: a dictionary containing the meta_rule
-
- Here is an example of a meta_rule:
- {
- "sub_meta_rules": {
- "relation_super": {
- "subject_categories": ["role"],
- "action_categories": ["computing_action"],
- "object_categories": ["id"],
- "relation": "relation_super"
- }
- },
- "aggregation": "and_true_aggregation"
- }
- """
+ def set_aggregation_algorithm(self, intra_extension_id, aggregation_algorithm_id, aggregation_algorithm_dict):
raise exception.NotImplemented() # pragma: no cover
- def set_sub_meta_rule_dict(self, extension_uuid, meta_rule_dict):
- """Set the Meta rule
+ def get_aggregation_algorithm(self, intra_extension_id):
+ raise exception.NotImplemented() # pragma: no cover
- :param extension_uuid: IntraExtension UUID
- :type extension_uuid: string
- :param meta_rule: a dictionary representing the meta_rule (see below)
- :return:a dictionary containing the meta_rule
+ def get_sub_meta_rules_dict(self, intra_extension_id):
+ raise exception.NotImplemented() # pragma: no cover
- Here is an example of a meta_rule:
- {
- "sub_meta_rules": {
- "relation_super": {
- "subject_categories": ["role"],
- "action_categories": ["computing_action"],
- "object_categories": ["id"],
- "relation": "relation_super"
- }
- },
- "aggregation": "and_true_aggregation"
- }
- """
+ def set_sub_meta_rule_dict(self, intra_extension_id, sub_meta_rule_id, meta_rule_dict):
+ raise exception.NotImplemented() # pragma: no cover
+
+ def del_sub_meta_rule(self, intra_extension_id, sub_meta_rule_id):
raise exception.NotImplemented() # pragma: no cover
# Rule functions
- def get_rule_dict(self, extension_uuid):
- """Get all rules
-
- :param extension_uuid: IntraExtension UUID
- :type extension_uuid: string
- :return: a dictionary containing rules ie.
- {
- "relation_super":[
- ["admin", "vm_admin", "servers", True],
- ["admin", "vm_access", "servers", True]
- ]
- }
- All items will be UUID.
- The last boolean item is the positive/negative value. If True, request that conforms to that rule
- will be authorized, if false, request will be rejected.
- """
+ def get_rules_dict(self, intra_extension_id, sub_meta_rule_id):
raise exception.NotImplemented() # pragma: no cover
- def set_rule_dict(self, extension_uuid, rule_dict):
- """Set all rules
-
- :param extension_uuid: IntraExtension UUID
- :type extension_uuid: string
- :param rules: a dictionary containing rules (see below)
- :type rules: dict
- :return: a dictionary containing rules ie.
- {
- "relation_super":[
- ["admin", "vm_admin", "servers", True],
- ["admin", "vm_access", "servers", True]
- ]
- }
- All items will be UUID.
- The last boolean item is the positive/negative value. If True, request that conforms to that rule
- will be authorized, if false, request will be rejected.
- """
+ def set_rule_dict(self, intra_extension_id, sub_meta_rule_id, rule_id, rule_list):
raise exception.NotImplemented() # pragma: no cover
+ def del_rule_dict(self, intra_extension_id, sub_meta_rule_id, rule_id):
+ raise exception.NotImplemented() # pragma: no cover
class LogDriver(object):
title = 'Extension Error'
-class IntraExtensionUnMapped(IntraExtensionException):
- message_format = _("The Extension is not mapped to a tenant.")
- code = 400
- title = 'Extension UUID Not Found Error'
- logger = "WARNING"
-
-
class IntraExtensionUnknown(IntraExtensionException):
message_format = _("The intra_extension is unknown.")
code = 400
logger = "ERROR"
-class AggregationAlgorithmExisting(AuthzMetadata):
- message_format = _("The given aggregation algorithm is existing.")
- code = 400
- title = 'Aggregation Algorithm Existing'
- logger = "ERROR"
-
-
class AggregationAlgorithmNotExisting(AuthzMetadata):
message_format = _("The given aggregation algorithm is not existing.")
code = 400
title = 'Sub Meta Rule Unknown'
logger = "ERROR"
+
class SubMetaRuleNameExisting(AuthzMetadata):
message_format = _("The sub meta rule name is existing.")
code = 400
path=self.PATH_PREFIX+'/intra_extensions/{intra_extension_id}/subject_categories/{subject_category_id}',
get_action='get_subject_category',
delete_action='del_subject_category',
+ post_action='set_subject_category',
rel=self._get_rel('subject_categories'),
path_vars={
'intra_extension_id': self._get_path('intra_extensions'),
path=self.PATH_PREFIX+'/intra_extensions/{intra_extension_id}/subjects/{subject_id}',
get_action='get_subject',
delete_action='del_subject',
+ post_action='set_subject',
rel=self._get_rel('subjects'),
path_vars={
'intra_extension_id': self._get_path('intra_extensions'),
path=self.PATH_PREFIX+'/intra_extensions/{intra_extension_id}/subject_scopes/{subject_category_id}/{subject_scope_id}',
get_action='get_subject_scope',
delete_action='del_subject_scope',
+ post_action='set_subject_scope',
rel=self._get_rel('subject_scope'),
path_vars={
'intra_extension_id': self._get_path('intra_extensions'),
'intra_extension_id': self._get_path('intra_extensions'),
})
- # Assignment route
- self._add_resource(
- mapper, intra_ext_controller,
- path=self.PATH_PREFIX+'/intra_extensions/{intra_extension_id}/subject_assignments/{subject_id}',
- get_action='get_subject_assignments',
- rel=self._get_rel('subject_assignments'),
- path_vars={
- 'intra_extension_id': self._get_path('intra_extensions'),
- })
self._add_resource(
mapper, intra_ext_controller,
path=self.PATH_PREFIX+'/intra_extensions/{intra_extension_id}/subject_assignments',
self._add_resource(
mapper, intra_ext_controller,
path=self.PATH_PREFIX+'/intra_extensions/{intra_extension_id}/aggregation_algorithm',
- post_action='add_aggregation_algorithm',
+ post_action='set_aggregation_algorithm',
rel=self._get_rel('aggregation_algorithms'),
path_vars={
'intra_extension_id': self._get_path('intra_extensions'),
mapper, intra_ext_controller,
path=self.PATH_PREFIX+'/intra_extensions/{intra_extension_id}/aggregation_algorithm/{aggregation_algorithm_id}',
get_action='get_aggregation_algorithm',
- delete_action='del_aggregation_algorithm',
rel=self._get_rel('aggregation_algorithms'),
path_vars={
'intra_extension_id': self._get_path('intra_extensions'),
path=self.PATH_PREFIX+'/intra_extensions/{intra_extension_id}/sub_meta_rules/{sub_meta_rule_id}',
get_action='get_sub_meta_rule',
delete_action='del_sub_meta_rule',
+ post_action='set_sub_meta_rule',
rel=self._get_rel('sub_meta_rules'),
path_vars={
'intra_extension_id': self._get_path('intra_extensions'),
path=self.PATH_PREFIX+'/intra_extensions/{intra_extension_id}/rule/{sub_meta_rule_id}/{rule_id}',
get_action='get_rule',
delete_action='del_rule',
+ post_action='set_rule',
rel=self._get_rel('rules'),
path_vars={
'intra_extension_id': self._get_path('intra_extensions'),
self.create_user("admin")
self.create_intra_extension()
- subjects = self.manager.get_subject_dict("admin", self.ref["id"])
+ subjects = self.manager.get_subjects_dict("admin", self.ref["id"])
self.assertIsInstance(subjects, dict)
self.assertIn("subjects", subjects)
self.assertIn("id", subjects)
# Delete the new subject
self.manager.del_subject("admin", self.ref["id"], new_subject["id"])
- subjects = self.manager.get_subject_dict("admin", self.ref["id"])
+ subjects = self.manager.get_subjects_dict("admin", self.ref["id"])
self.assertIsInstance(subjects, dict)
self.assertIn("subjects", subjects)
self.assertIn("id", subjects)
self.assertNotIn(new_subject["id"], subjects["subjects"])
# Add a particular subject
- subjects = self.manager.add_subject("admin", self.ref["id"], new_subject["id"])
+ subjects = self.manager.add_subject_dict("admin", self.ref["id"], new_subject["id"])
self.assertIsInstance(subjects, dict)
self.assertIn("subject", subjects)
self.assertIn("uuid", subjects["subject"])
self.assertEqual(new_subject["name"], subjects["subject"]["name"])
- subjects = self.manager.get_subject_dict("admin", self.ref["id"])
+ subjects = self.manager.get_subjects_dict("admin", self.ref["id"])
self.assertIsInstance(subjects, dict)
self.assertIn("subjects", subjects)
self.assertIn("id", subjects)
self.create_user("admin")
self.create_intra_extension()
- subject_categories = self.manager.get_subject_category_dict("admin", self.ref["id"])
+ subject_categories = self.manager.get_subject_categories_dict("admin", self.ref["id"])
self.assertIsInstance(subject_categories, dict)
self.assertIn("subject_categories", subject_categories)
self.assertIn("id", subject_categories)
# Delete the new subject_category
self.manager.del_subject_category("admin", self.ref["id"], new_subject_category["id"])
- subject_categories = self.manager.get_subject_category_dict("admin", self.ref["id"])
+ subject_categories = self.manager.get_subject_categories_dict("admin", self.ref["id"])
self.assertIsInstance(subject_categories, dict)
self.assertIn("subject_categories", subject_categories)
self.assertIn("id", subject_categories)
self.assertIn("uuid", subject_categories["subject_category"])
self.assertEqual(new_subject_category["name"], subject_categories["subject_category"]["name"])
new_subject_category["id"] = subject_categories["subject_category"]["uuid"]
- subject_categories = self.manager.get_subject_category_dict(
+ subject_categories = self.manager.get_subject_categories_dict(
"admin",
self.ref["id"])
self.assertIsInstance(subject_categories, dict)
)
for subject_category in subject_categories["subject_categories"]:
- subject_category_scope = self.manager.get_subject_scope_dict(
+ subject_category_scope = self.manager.get_subject_scopes_dict(
"admin",
self.ref["id"],
subject_category)
self.ref["id"],
subject_category,
new_subject_category_scope_uuid)
- subject_category_scope = self.manager.get_subject_scope_dict(
+ subject_category_scope = self.manager.get_subject_scopes_dict(
"admin",
self.ref["id"],
subject_category)
self.assertNotIn(new_subject_category_scope_uuid, subject_category_scope["subject_category_scope"])
# Add a particular subject_category_scope
- subject_category_scope = self.manager.add_subject_scope(
+ subject_category_scope = self.manager.add_subject_scope_dict(
"admin",
self.ref["id"],
subject_category,
self.assertIn("uuid", subject_category_scope["subject_category_scope"])
self.assertEqual(new_subject_category_scope[new_subject_category_scope_uuid],
subject_category_scope["subject_category_scope"]["name"])
- subject_category_scope = self.manager.get_subject_scope_dict(
+ subject_category_scope = self.manager.get_subject_scopes_dict(
"admin",
self.ref["id"],
subject_category)
)
for subject_category in subject_categories["subject_categories"]:
- subject_category_scope = self.manager.get_subject_scope_dict(
+ subject_category_scope = self.manager.get_subject_scopes_dict(
"admin",
self.ref["id"],
subject_category)
self.assertIn("and_true_aggregation", aggregation_algorithms["aggregation_algorithms"])
self.assertIn("test_aggregation", aggregation_algorithms["aggregation_algorithms"])
- aggregation_algorithm = self.manager.get_aggregation_algorithm("admin", self.ref["id"])
+ aggregation_algorithm = self.manager.get_aggregation_algorithm_dict("admin", self.ref["id"])
self.assertIsInstance(aggregation_algorithm, dict)
self.assertIn("aggregation", aggregation_algorithm)
self.assertIn(aggregation_algorithm["aggregation"], aggregation_algorithms["aggregation_algorithms"])
_aggregation_algorithm = list(aggregation_algorithms["aggregation_algorithms"])
_aggregation_algorithm.remove(aggregation_algorithm["aggregation"])
- aggregation_algorithm = self.manager.set_aggregation_algorithm("admin", self.ref["id"], _aggregation_algorithm[0])
+ aggregation_algorithm = self.manager.set_aggregation_algorithm_dict("admin", self.ref["id"], _aggregation_algorithm[0])
self.assertIsInstance(aggregation_algorithm, dict)
self.assertIn("aggregation", aggregation_algorithm)
self.assertIn(aggregation_algorithm["aggregation"], aggregation_algorithms["aggregation_algorithms"])
- sub_meta_rules = self.manager.get_sub_meta_rule_dict("admin", self.ref["id"])
+ sub_meta_rules = self.manager.get_sub_meta_rules_dict("admin", self.ref["id"])
self.assertIsInstance(sub_meta_rules, dict)
self.assertIn("sub_meta_rules", sub_meta_rules)
sub_meta_rules_conf = json.load(open(os.path.join(self.policy_directory, self.ref["model"], "metarule.json")))
metarule = dict()
categories = {
- "subject_categories": self.manager.get_subject_category_dict("admin", self.ref["id"]),
+ "subject_categories": self.manager.get_subject_categories_dict("admin", self.ref["id"]),
"object_categories": self.manager.get_object_category_dict("admin", self.ref["id"]),
"action_categories": self.manager.get_action_category_dict("admin", self.ref["id"])
}
self.ref["id"],
new_subject_category["name"])
new_subject_category["id"] = data["subject_category"]["uuid"]
- subject_categories = self.manager.get_subject_category_dict(
+ subject_categories = self.manager.get_subject_categories_dict(
"admin",
self.ref["id"])
self.assertIsInstance(subject_categories, dict)
self.create_user("admin")
self.create_intra_extension()
- sub_meta_rules = self.manager.get_sub_meta_rule_dict("admin", self.ref["id"])
+ sub_meta_rules = self.manager.get_sub_meta_rules_dict("admin", self.ref["id"])
self.assertIsInstance(sub_meta_rules, dict)
self.assertIn("sub_meta_rules", sub_meta_rules)
rules[relation] = list()
for rule in sub_rules["rules"][relation]:
for cat, cat_func, func_name in (
- ("subject_categories", self.manager.get_subject_scope_dict, "subject_category_scope"),
+ ("subject_categories", self.manager.get_subject_scopes_dict, "subject_category_scope"),
("action_categories", self.manager.get_action_scope_dict, "action_category_scope"),
("object_categories", self.manager.get_object_scope_dict, "object_category_scope"),
):
relation = sub_rules["rules"].keys()[0]
sub_rule = []
for cat, cat_func, func_name in (
- ("subject_categories", self.manager.get_subject_scope_dict, "subject_category_scope"),
+ ("subject_categories", self.manager.get_subject_scopes_dict, "subject_category_scope"),
("action_categories", self.manager.get_action_scope_dict, "action_category_scope"),
("object_categories", self.manager.get_object_scope_dict, "object_category_scope"),
):
rules[relation] = list()
for rule in sub_rules["rules"][relation]:
for cat, cat_func, func_name in (
- ("subject_categories", self.manager.get_subject_scope_dict, "subject_category_scope"),
+ ("subject_categories", self.manager.get_subject_scopes_dict, "subject_category_scope"),
("action_categories", self.manager.get_action_scope_dict, "action_category_scope"),
("object_categories", self.manager.get_object_scope_dict, "object_category_scope"),
):
self.assertRaises(
SubjectReadNotAuthorized,
- self.manager.get_subject_dict,
+ self.manager.get_subjects_dict,
demo_user["id"], ref["id"])
- subjects = self.manager.get_subject_dict(admin_user["id"], ref["id"])
+ subjects = self.manager.get_subjects_dict(admin_user["id"], ref["id"])
self.assertIsInstance(subjects, dict)
self.assertIn("subjects", subjects)
self.assertIn("id", subjects)
demo_user["id"], ref["id"], new_subject["id"])
self.manager.del_subject(admin_user["id"], ref["id"], new_subject["id"])
- subjects = self.manager.get_subject_dict(admin_user["id"], ref["id"])
+ subjects = self.manager.get_subjects_dict(admin_user["id"], ref["id"])
self.assertIsInstance(subjects, dict)
self.assertIn("subjects", subjects)
self.assertIn("id", subjects)
# Add a particular subject
self.assertRaises(
SubjectAddNotAuthorized,
- self.manager.add_subject,
+ self.manager.add_subject_dict,
demo_user["id"], ref["id"], new_subject["id"])
- subjects = self.manager.add_subject(admin_user["id"], ref["id"], new_subject["id"])
+ subjects = self.manager.add_subject_dict(admin_user["id"], ref["id"], new_subject["id"])
self.assertIsInstance(subjects, dict)
self.assertIn("subject", subjects)
self.assertIn("uuid", subjects["subject"])
self.assertEqual(new_subject["name"], subjects["subject"]["name"])
- subjects = self.manager.get_subject_dict(admin_user["id"], ref["id"])
+ subjects = self.manager.get_subjects_dict(admin_user["id"], ref["id"])
self.assertIsInstance(subjects, dict)
self.assertIn("subjects", subjects)
self.assertIn("id", subjects)
self.assertRaises(
SubjectCategoryReadNotAuthorized,
- self.manager.get_subject_category_dict,
+ self.manager.get_subject_categories_dict,
demo_user["id"], ref["id"])
- subject_categories = self.manager.get_subject_category_dict(admin_user["id"], ref["id"])
+ subject_categories = self.manager.get_subject_categories_dict(admin_user["id"], ref["id"])
self.assertIsInstance(subject_categories, dict)
self.assertIn("subject_categories", subject_categories)
self.assertIn("id", subject_categories)
demo_user["id"], ref["id"], new_subject_category["id"])
self.manager.del_subject_category(admin_user["id"], ref["id"], new_subject_category["id"])
- subject_categories = self.manager.get_subject_category_dict(admin_user["id"], ref["id"])
+ subject_categories = self.manager.get_subject_categories_dict(admin_user["id"], ref["id"])
self.assertIsInstance(subject_categories, dict)
self.assertIn("subject_categories", subject_categories)
self.assertIn("id", subject_categories)
self.assertIn("uuid", subject_categories["subject_category"])
self.assertEqual(new_subject_category["name"], subject_categories["subject_category"]["name"])
new_subject_category["id"] = subject_categories["subject_category"]["uuid"]
- subject_categories = self.manager.get_subject_category_dict(
+ subject_categories = self.manager.get_subject_categories_dict(
admin_user["id"],
ref["id"])
self.assertIsInstance(subject_categories, dict)
for subject_category in subject_categories["subject_categories"]:
self.assertRaises(
SubjectCategoryScopeReadNotAuthorized,
- self.manager.get_subject_scope_dict,
+ self.manager.get_subject_scopes_dict,
demo_user["id"], ref["id"], subject_category)
- subject_category_scope = self.manager.get_subject_scope_dict(
+ subject_category_scope = self.manager.get_subject_scopes_dict(
admin_user["id"],
ref["id"],
subject_category)
ref["id"],
subject_category,
new_subject_category_scope_uuid)
- subject_category_scope = self.manager.get_subject_scope_dict(
+ subject_category_scope = self.manager.get_subject_scopes_dict(
admin_user["id"],
ref["id"],
subject_category)
# Add a particular subject_category_scope
self.assertRaises(
SubjectCategoryScopeAddNotAuthorized,
- self.manager.add_subject_scope,
+ self.manager.add_subject_scope_dict,
demo_user["id"], ref["id"], subject_category,
new_subject_category_scope[new_subject_category_scope_uuid])
- subject_category_scope = self.manager.add_subject_scope(
+ subject_category_scope = self.manager.add_subject_scope_dict(
admin_user["id"],
ref["id"],
subject_category,
self.assertIn("uuid", subject_category_scope["subject_category_scope"])
self.assertEqual(new_subject_category_scope[new_subject_category_scope_uuid],
subject_category_scope["subject_category_scope"]["name"])
- subject_category_scope = self.manager.get_subject_scope_dict(
+ subject_category_scope = self.manager.get_subject_scopes_dict(
admin_user["id"],
ref["id"],
subject_category)
)
for subject_category in subject_categories["subject_categories"]:
- subject_category_scope = self.manager.get_subject_scope_dict(
+ subject_category_scope = self.manager.get_subject_scopes_dict(
admin_user["id"],
ref["id"],
subject_category)
self.assertRaises(
MetaRuleReadNotAuthorized,
- self.manager.get_aggregation_algorithm,
+ self.manager.get_aggregation_algorithm_dict,
demo_user["id"], ref["id"]
)
- aggregation_algorithm = self.manager.get_aggregation_algorithm(admin_user["id"], ref["id"])
+ aggregation_algorithm = self.manager.get_aggregation_algorithm_dict(admin_user["id"], ref["id"])
self.assertIsInstance(aggregation_algorithm, dict)
self.assertIn("aggregation", aggregation_algorithm)
self.assertIn(aggregation_algorithm["aggregation"], aggregation_algorithms["aggregation_algorithms"])
demo_user["id"], ref["id"], _aggregation_algorithm[0]
)
- aggregation_algorithm = self.manager.set_aggregation_algorithm(admin_user["id"], ref["id"], _aggregation_algorithm[0])
+ aggregation_algorithm = self.manager.set_aggregation_algorithm_dict(admin_user["id"], ref["id"], _aggregation_algorithm[0])
self.assertIsInstance(aggregation_algorithm, dict)
self.assertIn("aggregation", aggregation_algorithm)
self.assertIn(aggregation_algorithm["aggregation"], aggregation_algorithms["aggregation_algorithms"])
self.assertRaises(
MetaRuleReadNotAuthorized,
- self.manager.get_sub_meta_rule_dict,
+ self.manager.get_sub_meta_rules_dict,
demo_user["id"], ref["id"]
)
- sub_meta_rules = self.manager.get_sub_meta_rule_dict(admin_user["id"], ref["id"])
+ sub_meta_rules = self.manager.get_sub_meta_rules_dict(admin_user["id"], ref["id"])
self.assertIsInstance(sub_meta_rules, dict)
self.assertIn("sub_meta_rules", sub_meta_rules)
sub_meta_rules_conf = json.load(open(os.path.join(self.policy_directory, ref["model"], "metarule.json")))
metarule = dict()
categories = {
- "subject_categories": self.manager.get_subject_category_dict(admin_user["id"], ref["id"]),
+ "subject_categories": self.manager.get_subject_categories_dict(admin_user["id"], ref["id"]),
"object_categories": self.manager.get_object_category_dict(admin_user["id"], ref["id"]),
"action_categories": self.manager.get_action_category_dict(admin_user["id"], ref["id"])
}
ref["id"],
new_subject_category["name"])
new_subject_category["id"] = data["subject_category"]["uuid"]
- subject_categories = self.manager.get_subject_category_dict(
+ subject_categories = self.manager.get_subject_categories_dict(
admin_user["id"],
ref["id"])
self.assertIsInstance(subject_categories, dict)
self.assertRaises(
MetaRuleAddNotAuthorized,
- self.manager.get_sub_meta_rule,
+ self.manager.get_sub_meta_rule_dict,
demo_user["id"], ref["id"], metarule
)
- _sub_meta_rules = self.manager.get_sub_meta_rule(admin_user["id"], ref["id"], metarule)
+ _sub_meta_rules = self.manager.get_sub_meta_rule_dict(admin_user["id"], ref["id"], metarule)
self.assertIn(relation, metarule)
for item in ("subject_categories", "object_categories", "action_categories"):
self.assertEqual(
admin_user = self.create_user("admin")
ref = self.create_intra_extension()
- sub_meta_rules = self.manager.get_sub_meta_rule_dict(admin_user["id"], ref["id"])
+ sub_meta_rules = self.manager.get_sub_meta_rules_dict(admin_user["id"], ref["id"])
self.assertIsInstance(sub_meta_rules, dict)
self.assertIn("sub_meta_rules", sub_meta_rules)
rules[relation] = list()
for rule in sub_rules["rules"][relation]:
for cat, cat_func, func_name in (
- ("subject_categories", self.manager.get_subject_scope_dict, "subject_category_scope"),
+ ("subject_categories", self.manager.get_subject_scopes_dict, "subject_category_scope"),
("action_categories", self.manager.get_action_scope_dict, "action_category_scope"),
("object_categories", self.manager.get_object_scope_dict, "object_category_scope"),
):
relation = sub_rules["rules"].keys()[0]
sub_rule = []
for cat, cat_func, func_name in (
- ("subject_categories", self.manager.get_subject_scope_dict, "subject_category_scope"),
+ ("subject_categories", self.manager.get_subject_scopes_dict, "subject_category_scope"),
("action_categories", self.manager.get_action_scope_dict, "action_category_scope"),
("object_categories", self.manager.get_object_scope_dict, "object_category_scope"),
):
rules[relation] = list()
for rule in sub_rules["rules"][relation]:
for cat, cat_func, func_name in (
- ("subject_categories", self.manager.get_subject_scope_dict, "subject_category_scope"),
+ ("subject_categories", self.manager.get_subject_scopes_dict, "subject_category_scope"),
("action_categories", self.manager.get_action_scope_dict, "action_category_scope"),
("object_categories", self.manager.get_object_scope_dict, "object_category_scope"),
):
# Test when subject is known but not the object
demo_user = self.create_user("demo")
- self.manager.add_subject(
+ self.manager.add_subject_dict(
admin_user['id'],
self.ref["id"],
demo_user["id"]
)
my_subject_category = {"id": _tmp[0], "name": _tmp[1]}
- _tmp = self.manager.add_subject_scope(
+ _tmp = self.manager.add_subject_scope_dict(
admin_user['id'],
self.ref["id"],
my_subject_category["id"],
"relation": "relation_super"
}
}
- self.manager.get_sub_meta_rule(
+ self.manager.get_sub_meta_rule_dict(
admin_user['id'],
self.ref["id"],
my_meta_rule
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"])
+ subjects = self.manager.get_subjects_dict(admin_user["id"], tenant["id"])
self.assertIsInstance(subjects, dict)
self.assertIn("subjects", subjects)
self.assertIn("id", subjects)
# Add a particular subject
self.assertRaises(
SubjectAddNotAuthorized,
- self.manager.add_subject,
+ self.manager.add_subject_dict,
admin_user["id"], ref["id"], new_subject["id"])
def test_objects(self):
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"])
+ subject_categories = self.manager.get_subject_categories_dict(admin_user["id"], ref["id"])
self.assertIsInstance(subject_categories, dict)
self.assertIn("subject_categories", subject_categories)
self.assertIn("id", subject_categories)
)
for subject_category in subject_categories["subject_categories"]:
- subject_category_scope = self.manager.get_subject_scope_dict(
+ subject_category_scope = self.manager.get_subject_scopes_dict(
admin_user["id"],
ref["id"],
subject_category)
# Add a particular subject_category_scope
self.assertRaises(
SubjectCategoryScopeAddNotAuthorized,
- self.manager.add_subject_scope,
+ self.manager.add_subject_scope_dict,
admin_user["id"], ref["id"], subject_category, new_subject_category_scope[new_subject_category_scope_uuid])
def test_object_category_scope(self):
)
for subject_category in subject_categories["subject_categories"]:
- subject_category_scope = self.admin_manager.get_subject_scope_dict(
+ subject_category_scope = self.admin_manager.get_subject_scopes_dict(
admin_user["id"],
ref["id"],
subject_category)
self.assertIn("and_true_aggregation", aggregation_algorithms["aggregation_algorithms"])
self.assertIn("test_aggregation", aggregation_algorithms["aggregation_algorithms"])
- aggregation_algorithm = self.manager.get_aggregation_algorithm(admin_user["id"], ref["id"])
+ aggregation_algorithm = self.manager.get_aggregation_algorithm_dict(admin_user["id"], ref["id"])
self.assertIsInstance(aggregation_algorithm, dict)
self.assertIn("aggregation", aggregation_algorithm)
self.assertIn(aggregation_algorithm["aggregation"], aggregation_algorithms["aggregation_algorithms"])
_aggregation_algorithm.remove(aggregation_algorithm["aggregation"])
self.assertRaises(
MetaRuleAddNotAuthorized,
- self.manager.set_aggregation_algorithm,
+ self.manager.set_aggregation_algorithm_dict,
admin_user["id"], ref["id"], _aggregation_algorithm[0])
- sub_meta_rules = self.manager.get_sub_meta_rule_dict(admin_user["id"], ref["id"])
+ sub_meta_rules = self.manager.get_sub_meta_rules_dict(admin_user["id"], ref["id"])
self.assertIsInstance(sub_meta_rules, dict)
self.assertIn("sub_meta_rules", sub_meta_rules)
sub_meta_rules_conf = json.load(open(os.path.join(self.policy_directory, ref["model"], "metarule.json")))
metarule = dict()
categories = {
- "subject_categories": self.manager.get_subject_category_dict(admin_user["id"], ref["id"]),
+ "subject_categories": self.manager.get_subject_categories_dict(admin_user["id"], ref["id"]),
"object_categories": self.manager.get_object_category_dict(admin_user["id"], ref["id"]),
"action_categories": self.manager.get_action_category_dict(admin_user["id"], ref["id"])
}
ref["id"],
new_subject_category["name"])
new_subject_category["id"] = data["subject_category"]["uuid"]
- subject_categories = self.manager.get_subject_category_dict(
+ subject_categories = self.manager.get_subject_categories_dict(
admin_user["id"],
ref["id"])
self.assertIsInstance(subject_categories, dict)
metarule[relation]["subject_categories"].append(new_subject_category["id"])
self.assertRaises(
MetaRuleAddNotAuthorized,
- self.manager.get_sub_meta_rule,
+ self.manager.get_sub_meta_rule_dict,
admin_user["id"], ref["id"], metarule)
def test_sub_rules(self):
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_dict(admin_user["id"], ref["id"])
+ sub_meta_rules = self.manager.get_sub_meta_rules_dict(admin_user["id"], ref["id"])
self.assertIsInstance(sub_meta_rules, dict)
self.assertIn("sub_meta_rules", sub_meta_rules)
rules[relation] = list()
for rule in sub_rules["rules"][relation]:
for cat, cat_func, func_name in (
- ("subject_categories", self.manager.get_subject_scope_dict, "subject_category_scope"),
+ ("subject_categories", self.manager.get_subject_scopes_dict, "subject_category_scope"),
("action_categories", self.manager.get_action_scope_dict, "action_category_scope"),
("object_categories", self.manager.get_object_scope_dict, "object_category_scope"),
):
relation = sub_rules["rules"].keys()[0]
sub_rule = []
for cat, cat_func, func_name in (
- ("subject_categories", self.manager.get_subject_scope_dict, "subject_category_scope"),
+ ("subject_categories", self.manager.get_subject_scopes_dict, "subject_category_scope"),
("action_categories", self.manager.get_action_scope_dict, "action_category_scope"),
("object_categories", self.manager.get_object_scope_dict, "object_category_scope"),
):
Code Review / moon.git / commitdiff