# -*- coding: utf-8 -*- """DM11 RECEIPT VALIDATOR - an independent checker. **This single file is the whole tool.** It imports nothing but the Python standard library and reads no DM11 code, so a DM11 receipt can be checked without trusting anything DM11 wrote. That is what "independent" means here. python dm11_validate.py receipt.json python dm11_validate.py receipt.json --json What it reports: SCHEMA whether the expected fields are present and the version known INTEGRITY whether the hash matches the contents PROOF whether evidence supporting the verdict is present UNKNOWNS what the receipt marks as unmeasured or unknown (**kept distinct from a field that is simply empty**) LIMITATIONS what the issuer declared it did not measure **Three values, never two.** PASS / FAIL / NOT_CHECKED are not merged. Something that could not be checked is not reported as PASS, and not as FAIL. **A hash does not prove the business reached the state the record describes.** It says only that the record was not altered after it was issued. When this validator prints INTEGRITY PASS, it has not said the business outcome happened. Exit codes: 0 no FAIL (NOT_CHECKED results may still be present) 1 one or more FAIL 2 could not read the input at all (missing file, or not JSON) """ from __future__ import annotations import hashlib import json import sys VALIDATOR_VERSION = "1" # Receipt versions this validator knows. **An unknown version is # NOT_CHECKED, never PASS.** # Being independent means this file cannot read DM11's own schema module, # so the list is written by hand. That is exactly why an unknown version # is NOT_CHECKED rather than FAIL: this list may simply be out of date. KNOWN_RECEIPT_VERSIONS = ("2.1", "2.0", "RR-2.0", "2", "V2") KNOWN_SCHEMA_IDS = ("dm11.reality_receipt",) # The fields a receipt is expected to carry. REQUIRED_FIELDS = ( "receipt_version", "transaction_id", "intent", "authority", "plan_hash", "before_state_ref", "preconditions", "actual_steps", "after_state_ref", "postcondition_results", "action_status", "settlement_status", "outcome_status", "final_status", "data_origin", "issued_at", "limitations", "receipt_hash", ) # Words that mean "not measured". **Counted separately from an empty # field** - a field nobody filled in is not the same as a field that says # the thing was not measured. UNKNOWN_WORDS = ("UNKNOWN", "NOT_EVALUATED", "NOT_MEASURED", "NOT_CHECKED", "NOT_APPLICABLE", "OUTCOME_UNKNOWN", "OUTCOME_PENDING") # A receipt must say whether its data is synthetic or real. ORIGIN_SYNTHETIC = ("SYNTHETIC", "SYNTHETIC_DEMO", "MOCK", "SIMULATED") ORIGIN_REAL = ("REAL", "PRODUCTION", "IMPORTED", "LIVE") def _canon(o) -> str: return json.dumps(o, ensure_ascii=False, sort_keys=True, separators=(",", ":")) def _h(s: str) -> str: return hashlib.sha256(s.encode("utf-8")).hexdigest() def _r(section, name, verdict, detail=""): return {"section": section, "check": name, "verdict": verdict, "detail": detail} def check_schema(d: dict) -> list: out = [] missing = [f for f in REQUIRED_FIELDS if f not in d] out.append(_r("SCHEMA", "required_fields", "FAIL" if missing else "PASS", ("%d field(s) missing: %s" % (len(missing), ", ".join(missing))) if missing else "all %d fields present" % len(REQUIRED_FIELDS))) v = str(d.get("receipt_version", "")) if not v: out.append(_r("SCHEMA", "receipt_version", "FAIL", "no version")) elif v.upper() in KNOWN_RECEIPT_VERSIONS: out.append(_r("SCHEMA", "receipt_version", "PASS", "version %s is known" % v)) else: # An unknown version is not a failure of the receipt. It may only # mean this validator is older than the receipt. out.append(_r("SCHEMA", "receipt_version", "NOT_CHECKED", "version %s is unknown to this validator (v%s) - it cannot vouch for what the fields mean" % (v, VALIDATOR_VERSION))) sid = str(d.get("schema_id", "")) if not sid: out.append(_r("SCHEMA", "schema_id", "NOT_CHECKED", "no schema_id (a receipt older than 2.1)")) elif sid in KNOWN_SCHEMA_IDS: out.append(_r("SCHEMA", "schema_id", "PASS", sid)) else: out.append(_r("SCHEMA", "schema_id", "NOT_CHECKED", "%r is not a DM11 receipt - these checks do not apply to it" % sid)) o = str(d.get("data_origin", "")).upper() if not o: out.append(_r("SCHEMA", "data_origin", "FAIL", "does not say whether the data is synthetic or real")) elif any(o.startswith(x) for x in ORIGIN_SYNTHETIC): out.append(_r("SCHEMA", "data_origin", "PASS", "%s - **synthetic**. Not a record of a business event that happened." % o)) elif o in ORIGIN_REAL: out.append(_r("SCHEMA", "data_origin", "PASS", "%s - real data" % o)) else: out.append(_r("SCHEMA", "data_origin", "NOT_CHECKED", "%r is neither synthetic nor real" % d.get("data_origin"))) return out def check_integrity(d: dict) -> list: out = [] got = d.get("receipt_hash") if not got: return [_r("INTEGRITY", "receipt_hash", "NOT_CHECKED", "no hash")] body = {k: v for k, v in d.items() if k != "receipt_hash"} want = _h(_canon(body)) if got == want: out.append(_r("INTEGRITY", "receipt_hash", "PASS", "matches the contents (%s...). **This does not say the business reached that state.**" % str(got)[:16])) else: out.append(_r("INTEGRITY", "receipt_hash", "FAIL", "does not match - altered after it was issued. expected %s... / recorded %s..." % (want[:16], str(got)[:16]))) tl = d.get("trail_length") th = d.get("trail_head") if tl is None and th is None: out.append(_r("INTEGRITY", "trail", "NOT_CHECKED", "no audit trail recorded")) elif isinstance(tl, int) and tl > 0 and th: out.append(_r("INTEGRITY", "trail", "PASS", "a %d-entry trail, head %s... (checking the trail itself needs the original log)" % (tl, str(th)[:12]))) else: out.append(_r("INTEGRITY", "trail", "FAIL", "trail length %r and head %r do not agree" % (tl, th))) return out def check_proof(d: dict) -> list: "Whether evidence supporting the verdict is present. **Present / absent / not looked at are kept apart.**" out = [] final = str(d.get("final_status", "")) post = d.get("postcondition_results") if post is None: out.append(_r("PROOF", "postconditions", "NOT_CHECKED", "field absent")) elif not post: out.append(_r("PROOF", "postconditions", "FAIL" if final == "VERIFIED" else "NOT_CHECKED", "0 - **zero postconditions is not the same as all of them met**" + (", yet it calls itself VERIFIED" if final == "VERIFIED" else ""))) else: bad = [p for p in post if isinstance(p, dict) and p.get("ok") is not True] if final == "VERIFIED" and bad: out.append(_r("PROOF", "postconditions", "FAIL", "VERIFIED, yet %d postcondition(s) are not met" % len(bad))) else: out.append(_r("PROOF", "postconditions", "PASS", "%d checked (%d not met)" % (len(post), len(bad)))) # Does the receipt itself state the difference between its hash and # business truth? A reader who is not told will conflate them. note = " ".join(str(d.get(k, "")) for k in ("integrity_note", "causality_note")) out.append(_r("PROOF", "no_truth_overclaim", "PASS" if note.strip() else "FAIL", "the receipt states the difference between its hash and business truth" if note.strip() else "the difference is not stated - a reader may take the hash for the truth")) # The assurance profile. a = d.get("assurance") if a is None: out.append(_r("PROOF", "assurance", "NOT_CHECKED", "field absent (an older version)")) elif str(a.get("status", "")) == "NOT_EVALUATED": out.append(_r("PROOF", "assurance", "FAIL" if final == "VERIFIED" else "NOT_CHECKED", "the assurance level was never evaluated (profile %s)" % a.get("profile"))) elif a.get("missing"): out.append(_r("PROOF", "assurance", "FAIL" if final == "VERIFIED" else "PASS", "profile %s is missing: %s" % (a.get("profile"), ", ".join(map(str, a["missing"]))))) else: out.append(_r("PROOF", "assurance", "PASS", "profile %s is satisfied" % a.get("profile"))) # Executed steps. # A receipt that does not say what it left unmeasured misleads its # reader, who will assume everything was measured. lim = d.get("limitations") if lim is None: out.append(_r("PROOF", "limitations", "FAIL", "field absent")) elif not lim: out.append(_r("PROOF", "limitations", "FAIL", "0 - **the receipt does not say what it did not " "measure**. A reader will assume everything was " "measured.")) else: out.append(_r("PROOF", "limitations", "PASS", "%d recorded" % len(lim))) steps = d.get("actual_steps") if not steps: out.append(_r("PROOF", "actual_steps", "FAIL" if final in ("VERIFIED", "RECOVERED") else "NOT_CHECKED", "0 executed steps recorded")) else: out.append(_r("PROOF", "actual_steps", "PASS", "%d recorded" % len(steps))) return out def find_unknowns(d: dict) -> list: "Collect what the receipt marks as unmeasured. **Not the same as a field left empty.**" hits = [] def walk(o, path): if isinstance(o, dict): for k, v in o.items(): walk(v, "%s.%s" % (path, k)) elif isinstance(o, list): for i, v in enumerate(o): walk(v, "%s[%d]" % (path, i)) elif isinstance(o, str) and o.strip().upper() in UNKNOWN_WORDS: hits.append({"path": path.lstrip("."), "value": o}) walk(d, "") return hits def validate(d: dict) -> dict: checks = check_schema(d) + check_integrity(d) + check_proof(d) unk = find_unknowns(d) lim = d.get("limitations") or [] n = {"PASS": 0, "FAIL": 0, "NOT_CHECKED": 0} for c in checks: n[c["verdict"]] = n.get(c["verdict"], 0) + 1 return {"validator_version": VALIDATOR_VERSION, "transaction_id": d.get("transaction_id"), "checks": checks, "counts": n, "unknowns": unk, "limitations": list(lim), "verdict": "FAIL" if n["FAIL"] else "NO_FAILURES", # Never the word "passed". No failures found and the business # being correct are different statements. The distinction is # carried in a field, so rewording cannot erase it. "means": { "no_failures_implies_business_correct": False, "no_failures_implies_receipt_unaltered": not n["FAIL"] and any( c["check"] == "receipt_hash" and c["verdict"] == "PASS" for c in checks), "checks_not_performed": n["NOT_CHECKED"], }, "note": ("FAIL %d / PASS %d / could not check %d. FAIL 0 means " "**no problem within what this validator can see** - it " "does not say the business outcome was correct." % (n["FAIL"], n["PASS"], n["NOT_CHECKED"]))} def render(v: dict) -> str: L = ["DM11 RECEIPT VALIDATOR v%s" % v["validator_version"], "transaction %s" % v["transaction_id"], ""] cur = None mark = {"PASS": " OK ", "FAIL": " !! ", "NOT_CHECKED": " ?? "} for c in v["checks"]: if c["section"] != cur: cur = c["section"] L.append("[%s]" % cur) L.append("%s%-22s %s" % (mark[c["verdict"]], c["check"], c["detail"])) L.append("") L.append("[UNKNOWNS] marked unmeasured or unknown by the issuer: %d" % len(v["unknowns"])) for u in v["unknowns"][:12]: L.append(" %s = %s" % (u["path"], u["value"])) if len(v["unknowns"]) > 12: L.append(" ...and %d more" % (len(v["unknowns"]) - 12)) L.append("") L.append("[LIMITATIONS] declared by the issuer: %d" % len(v["limitations"])) for x in v["limitations"]: L.append(" - %s" % x) L.append("") L.append(v["note"]) return "\n".join(L) def main(argv): args = [a for a in argv[1:] if not a.startswith("--")] # A receipt's own text can be in any language. A console that cannot # encode it must not make this tool fail - it degrades the characters # it cannot print, and still reports the verdict. for st in (sys.stdout, sys.stderr): try: st.reconfigure(encoding="utf-8") except Exception: # noqa: BLE001 try: st.reconfigure(errors="backslashreplace") except Exception: # noqa: BLE001 pass as_json = "--json" in argv if not args: sys.stderr.write(__doc__ or "") return 2 try: with open(args[0], "r", encoding="utf-8-sig") as f: d = json.load(f) except Exception as e: # noqa: BLE001 sys.stderr.write("unreadable: %s\n" % e) return 2 if not isinstance(d, dict): sys.stderr.write("a receipt must be a JSON object (got: %s)\n" % type(d).__name__) return 2 v = validate(d) if as_json: sys.stdout.write(json.dumps(v, ensure_ascii=False, indent=2) + "\n") else: sys.stdout.write(render(v) + "\n") return 1 if v["counts"]["FAIL"] else 0 if __name__ == "__main__": sys.exit(main(sys.argv))