356 lines
12 KiB
Python
356 lines
12 KiB
Python
__author__ = "Roland Hedberg"
|
|
__license__ = "Apache 2.0"
|
|
__version__ = ""
|
|
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class PolicyError(Exception):
|
|
pass
|
|
|
|
|
|
def combine_subset_of(s1, s2): # pragma: no cover
|
|
return list(set(s1).intersection(set(s2)))
|
|
|
|
|
|
def combine_superset_of(s1, s2): # pragma: no cover
|
|
return list(set(s1).intersection(set(s2)))
|
|
|
|
|
|
def combine_one_of(s1, s2): # pragma: no cover
|
|
return list(set(s1).intersection(set(s2)))
|
|
|
|
|
|
def combine_add(s1, s2): # pragma: no cover
|
|
if isinstance(s1, list):
|
|
set1 = set(s1)
|
|
else:
|
|
set1 = {s1}
|
|
if isinstance(s2, list):
|
|
set2 = set(s2)
|
|
else:
|
|
set2 = {s2}
|
|
return list(set1.union(set2))
|
|
|
|
|
|
POLICY_FUNCTIONS = {
|
|
"subset_of",
|
|
"superset_of",
|
|
"one_of",
|
|
"add",
|
|
"value",
|
|
"default",
|
|
"essential",
|
|
}
|
|
|
|
OP2FUNC = {
|
|
"subset_of": combine_subset_of,
|
|
"superset_of": combine_superset_of,
|
|
"one_of": combine_one_of,
|
|
"add": combine_add,
|
|
}
|
|
|
|
|
|
def do_sub_one_super_add(superior, child, policy): # pragma: no cover
|
|
if policy in superior and policy in child:
|
|
comb = OP2FUNC[policy](superior[policy], child[policy])
|
|
if comb:
|
|
return comb
|
|
else:
|
|
raise PolicyError("Value sets doesn't overlap")
|
|
elif policy in superior:
|
|
return superior[policy]
|
|
elif policy in child:
|
|
return child[policy]
|
|
|
|
|
|
def do_value(superior, child, policy): # pragma: no cover
|
|
if policy in superior and policy in child:
|
|
if superior[policy] == child[policy]:
|
|
return superior[policy]
|
|
else:
|
|
raise PolicyError("Not allowed to combine values")
|
|
elif policy in superior:
|
|
return superior[policy]
|
|
elif policy in child:
|
|
return child[policy]
|
|
|
|
|
|
def do_default(superior, child, policy): # pragma: no cover
|
|
# A child's default can not override a superiors
|
|
if policy in superior and policy in child:
|
|
if superior["default"] == child["default"]:
|
|
return superior["default"]
|
|
else:
|
|
raise PolicyError("Not allowed to change default")
|
|
elif policy in superior:
|
|
return superior[policy]
|
|
elif policy in child:
|
|
return child[policy]
|
|
|
|
|
|
def do_essential(superior, child, policy): # pragma: no cover
|
|
# essential: an child can make it True if a superior has states False
|
|
# but not the other way around
|
|
|
|
if policy in superior and policy in child:
|
|
if not superior[policy] and child["essential"]:
|
|
return True
|
|
else:
|
|
return superior[policy]
|
|
elif policy in superior:
|
|
return superior[policy]
|
|
elif policy in child: # Not in superior is the same as essential=True
|
|
return True
|
|
|
|
|
|
DO_POLICY = {
|
|
"superset_of": do_sub_one_super_add,
|
|
"subset_of": do_sub_one_super_add,
|
|
"one_of": do_sub_one_super_add,
|
|
"add": do_sub_one_super_add,
|
|
"value": do_value,
|
|
"default": do_default,
|
|
"essential": do_essential,
|
|
}
|
|
|
|
|
|
def combine_claim_policy(superior, child): # noqa: C901 # pragma: no cover
|
|
"""
|
|
Combine policy rules.
|
|
Applying the child policy can only make the combined policy more restrictive.
|
|
:param superior: Superior policy
|
|
:param child: Intermediates policy
|
|
"""
|
|
# weed out everything I don't recognize
|
|
|
|
superior_set = set(superior).intersection(POLICY_FUNCTIONS)
|
|
child_set = set(child).intersection(POLICY_FUNCTIONS)
|
|
if "value" in superior_set: # An exact value can not be restricted.
|
|
if child_set:
|
|
if "essential" in child_set:
|
|
if len(child_set) == 1:
|
|
return {"value": superior["value"], "essential": child["essential"]}
|
|
else:
|
|
raise PolicyError(
|
|
"value can only be combined with essential, not {}".format(
|
|
child_set
|
|
)
|
|
)
|
|
elif "value" in child_set:
|
|
if child["value"] != superior["value"]: # Not OK
|
|
raise PolicyError("Child can not set another value then superior")
|
|
else:
|
|
return superior
|
|
else:
|
|
raise PolicyError(
|
|
"Not allowed combination of policies: {} + {}".format(
|
|
superior, child
|
|
)
|
|
)
|
|
return superior
|
|
else:
|
|
if "essential" in superior_set and "essential" in child_set:
|
|
# can only go from False to True
|
|
if (
|
|
superior["essential"] != child["essential"]
|
|
and child["essential"] is False
|
|
):
|
|
raise PolicyError("Essential can not go from True to False")
|
|
|
|
comb_policy = superior_set.union(child_set)
|
|
if "one_of" in comb_policy:
|
|
if "subset_of" in comb_policy or "superset_of" in comb_policy:
|
|
raise PolicyError(
|
|
"one_of can not be combined with subset_of/superset_of"
|
|
)
|
|
|
|
rule = {}
|
|
for policy in comb_policy:
|
|
rule[policy] = DO_POLICY[policy](superior, child, policy)
|
|
|
|
if comb_policy == {"superset_of", "subset_of"}:
|
|
# make sure the subset_of is a superset of superset_of.
|
|
if set(rule["superset_of"]).difference(set(rule["subset_of"])):
|
|
raise PolicyError("superset_of not a super set of subset_of")
|
|
elif comb_policy == {"superset_of", "subset_of", "default"}:
|
|
# make sure the subset_of is a superset of superset_of.
|
|
if set(rule["superset_of"]).difference(set(rule["subset_of"])):
|
|
raise PolicyError("superset_of not a super set of subset_of")
|
|
if set(rule["default"]).difference(set(rule["subset_of"])):
|
|
raise PolicyError("default not a sub set of subset_of")
|
|
if set(rule["superset_of"]).difference(set(rule["default"])):
|
|
raise PolicyError("default not a super set of subset_of")
|
|
elif comb_policy == {"subset_of", "default"}:
|
|
if set(rule["default"]).difference(set(rule["subset_of"])):
|
|
raise PolicyError("default not a sub set of subset_of")
|
|
elif comb_policy == {"superset_of", "default"}:
|
|
if set(rule["superset_of"]).difference(set(rule["default"])):
|
|
raise PolicyError("default not a super set of subset_of")
|
|
elif comb_policy == {"one_of", "default"}:
|
|
if isinstance(rule["default"], list):
|
|
if set(rule["default"]).difference(set(rule["one_of"])):
|
|
raise PolicyError("default not a super set of one_of")
|
|
else:
|
|
if {rule["default"]}.difference(set(rule["one_of"])):
|
|
raise PolicyError("default not a super set of one_of")
|
|
return rule
|
|
|
|
|
|
def combine_policy(superior, child):
|
|
res = {}
|
|
sup_set = set(superior.keys())
|
|
chi_set = set(child.keys())
|
|
|
|
for claim in set(sup_set).intersection(chi_set):
|
|
res[claim] = combine_claim_policy(superior[claim], child[claim])
|
|
|
|
for claim in sup_set.difference(chi_set):
|
|
res[claim] = superior[claim]
|
|
|
|
for claim in chi_set.difference(sup_set):
|
|
res[claim] = child[claim]
|
|
|
|
return res
|
|
|
|
|
|
def gather_policies(chain, entity_type):
|
|
"""
|
|
Gather and combine all the metadata policies that are defined in the trust chain
|
|
:param chain: A list of Entity Statements
|
|
:return: The combined metadata policy
|
|
"""
|
|
|
|
try:
|
|
combined_policy = chain[0]["metadata_policy"][entity_type]
|
|
except KeyError:
|
|
combined_policy = {}
|
|
|
|
for es in chain[1:]:
|
|
try:
|
|
child = es["metadata_policy"][entity_type]
|
|
except KeyError:
|
|
pass
|
|
else:
|
|
combined_policy = combine_policy(combined_policy, child)
|
|
|
|
return combined_policy
|
|
|
|
|
|
def union(val1, val2):
|
|
if isinstance(val1, list):
|
|
base = set(val1)
|
|
else:
|
|
base = {val1}
|
|
|
|
if isinstance(val2, list):
|
|
ext = set(val2)
|
|
else:
|
|
ext = {val2}
|
|
return base.union(ext)
|
|
|
|
|
|
def apply_policy(metadata, policy):
|
|
"""
|
|
Apply a metadata policy to a metadata statement.
|
|
The order is value, add, default and then the checks subset_of/superset_of and one_of
|
|
:param metadata: A metadata statement
|
|
:param policy: A metadata policy
|
|
:return: A metadata statement that adheres to a metadata policy
|
|
"""
|
|
metadata_set = set(metadata.keys())
|
|
policy_set = set(policy.keys())
|
|
|
|
# Metadata claims that there exists a policy for
|
|
for claim in metadata_set.intersection(policy_set):
|
|
if "value" in policy[claim]: # value overrides everything
|
|
metadata[claim] = policy[claim]["value"]
|
|
else:
|
|
if "one_of" in policy[claim]:
|
|
# The is for claims that can have only one value
|
|
if isinstance(metadata[claim], list): # Should not be but ...
|
|
_claim = [
|
|
c for c in metadata[claim] if c in policy[claim]["one_of"]
|
|
]
|
|
if _claim:
|
|
metadata[claim] = _claim[0]
|
|
else:
|
|
raise PolicyError(
|
|
"{}: None of {} among {}".format(
|
|
claim, metadata[claim], policy[claim]["one_of"]
|
|
)
|
|
)
|
|
else:
|
|
if metadata[claim] in policy[claim]["one_of"]:
|
|
pass
|
|
else:
|
|
raise PolicyError(
|
|
"{} not among {}".format(
|
|
metadata[claim], policy[claim]["one_of"]
|
|
)
|
|
)
|
|
else:
|
|
# The following is for claims that can have lists of values
|
|
if "add" in policy[claim]:
|
|
metadata[claim] = list(union(metadata[claim], policy[claim]["add"]))
|
|
|
|
if "subset_of" in policy[claim]:
|
|
_val = set(policy[claim]["subset_of"]).intersection(
|
|
set(metadata[claim])
|
|
)
|
|
if _val:
|
|
metadata[claim] = list(_val)
|
|
else:
|
|
raise PolicyError(
|
|
"{} not subset of {}".format(
|
|
metadata[claim], policy[claim]["subset_of"]
|
|
)
|
|
)
|
|
if "superset_of" in policy[claim]:
|
|
if set(policy[claim]["superset_of"]).difference(
|
|
set(metadata[claim])
|
|
):
|
|
raise PolicyError(
|
|
"{} not superset of {}".format(
|
|
metadata[claim], policy[claim]["superset_of"]
|
|
)
|
|
)
|
|
else:
|
|
pass
|
|
|
|
# In policy but not in metadata
|
|
for claim in policy_set.difference(metadata_set):
|
|
if "value" in policy[claim]:
|
|
metadata[claim] = policy[claim]["value"]
|
|
elif "add" in policy[claim]:
|
|
metadata[claim] = policy[claim]["add"]
|
|
elif "default" in policy[claim]:
|
|
metadata[claim] = policy[claim]["default"]
|
|
|
|
if claim not in metadata:
|
|
if "essential" in policy[claim] and policy[claim]["essential"]:
|
|
raise PolicyError("Essential claim '{}' missing".format(claim))
|
|
|
|
# All that are in metadata but not in policy should just remain
|
|
|
|
return metadata
|
|
|
|
|
|
def diff2policy(new, old):
|
|
res = {}
|
|
for claim in set(new).intersection(set(old)):
|
|
if new[claim] == old[claim]:
|
|
continue
|
|
else:
|
|
res[claim] = {"value": new[claim]}
|
|
|
|
for claim in set(new).difference(set(old)):
|
|
if claim in ["contacts"]:
|
|
res[claim] = {"add": new[claim]}
|
|
else:
|
|
res[claim] = {"value": new[claim]}
|
|
|
|
return res
|