107 lines
3.8 KiB
Python
107 lines
3.8 KiB
Python
"""Rights records, policies, stage checks and backend conditions.
|
|
|
|
Non-normative; see spec/14-rights.md.
|
|
"""
|
|
|
|
USES = ("read", "tdm", "redistribute", "quote-public", "accessible-copy")
|
|
PURPOSES = ("internal", "external")
|
|
PROCESSORS = ("local", "tenancy", "outside-no-retention", "outside")
|
|
AI_PAIRS = tuple(p + "/" + q for p in PURPOSES for q in PROCESSORS)
|
|
|
|
|
|
def _value(v):
|
|
return v if v in (True, False) else "unknown"
|
|
|
|
|
|
def flat(rights):
|
|
"""A rights record's uses as one flat map: the five uses plus ai/<purpose>/<processor>."""
|
|
uses = rights.get("uses", {})
|
|
out = {u: _value(uses.get(u, "unknown")) for u in USES}
|
|
ai = uses.get("ai", {})
|
|
for pair in AI_PAIRS:
|
|
out["ai/" + pair] = _value(ai.get(pair, "unknown"))
|
|
return out
|
|
|
|
|
|
def _policy_applies(policy, witness, now):
|
|
if policy.get("valid_until") and now >= policy["valid_until"]:
|
|
return False
|
|
if policy.get("at") and now < policy["at"]:
|
|
return False
|
|
a = policy.get("applies_to", {})
|
|
r = witness["rights"]
|
|
if "basis" in a and r.get("basis") not in a["basis"]:
|
|
return False
|
|
if "license" in a and (r.get("license") or {}).get("spdx") not in a["license"]:
|
|
return False
|
|
if "doi_prefixes" in a:
|
|
doi = (witness.get("origin") or {}).get("doi") or ""
|
|
if not any(doi.startswith(p) for p in a["doi_prefixes"]):
|
|
return False
|
|
if "witnesses" in a and witness["sha256"] not in a["witnesses"]:
|
|
return False
|
|
return True
|
|
|
|
|
|
def effective(witness, policies, now):
|
|
"""The witness's uses at `now`, with unknowns resolved by applicable policies, or all false if expired."""
|
|
r = witness["rights"]
|
|
f = flat(r)
|
|
if r.get("valid_until") and now >= r["valid_until"]:
|
|
return {k: False for k in f}, True
|
|
for p in policies:
|
|
if not _policy_applies(p, witness, now):
|
|
continue
|
|
pf = flat({"uses": p.get("uses", {})})
|
|
for k, v in pf.items():
|
|
if f[k] == "unknown" and v != "unknown":
|
|
f[k] = v
|
|
return f, False
|
|
|
|
|
|
def combine(flats):
|
|
"""The most restrictive of several witnesses' uses: false over unknown over true."""
|
|
out = {}
|
|
for k in flats[0]:
|
|
vals = [f[k] for f in flats]
|
|
out[k] = False if False in vals else ("unknown" if "unknown" in vals else True)
|
|
return out
|
|
|
|
|
|
def check(stage, processor, uses):
|
|
"""Whether a stage may run on a processor under `uses`. Returns (allowed, lacking)."""
|
|
lacking = []
|
|
for u in stage.get("uses", []):
|
|
if uses[u] is not True:
|
|
lacking.append(u if uses[u] is False else u + "?")
|
|
ai = stage.get("ai")
|
|
if ai:
|
|
allowed_here = ai.get("processors")
|
|
if allowed_here and processor not in allowed_here:
|
|
lacking.append("processor:" + processor)
|
|
k = "ai/" + ai["purpose"] + "/" + processor
|
|
if uses[k] is not True:
|
|
lacking.append(k if uses[k] is False else k + "?")
|
|
return (not lacking), lacking
|
|
|
|
|
|
def backend_allowed(backend, profile):
|
|
"""Whether a team profile meets every condition a backend's licenses impose. Returns (ok, failed)."""
|
|
failed = []
|
|
for cond in backend.get("conditions", []):
|
|
k = cond["kind"]
|
|
if k == "revenue-cap":
|
|
r = profile.get("revenue_or_funding_usd")
|
|
if r is None or r.get("at_most") is None or r["at_most"] > cond["usd"]:
|
|
failed.append(cond["id"])
|
|
elif k == "no-competition":
|
|
comp = profile.get("competes_with")
|
|
if comp is None or cond["vendor"] in comp:
|
|
failed.append(cond["id"])
|
|
elif k == "purpose":
|
|
if profile.get("purpose") not in cond["allowed"]:
|
|
failed.append(cond["id"])
|
|
else:
|
|
failed.append(cond["id"])
|
|
return (not failed), failed
|