#!/usr/bin/env python3 # SPDX-License-Identifier: Apache-2.0 """ADCR reference verifier, rulebook 0.1.0. Checks an export offline and reports a verdict and findings as SPEC.md EXP-04 defines them, for every conformance class: Record, Law, Chain, Anchor, Oversight, Rights and Export. Python standard library only, plus an OpenSSL 1.1 or later binary for one step: the signature of each RFC 3161 timestamp token is checked with `openssl ts -verify`, the reference RFC 3161 verifier. Parsing the token, its status, its message imprint and its signing certificate is done here. Without a usable openssl the signature and trust steps are reported as `anchor_token_not_checked` (verdict `unverified`), never as a failure: a tool that could not check must not accuse a genuine record. Usage: python3 -I adcr_verify.py export.json --trust-anchors anchors.pem python3 -I adcr_verify.py export.json --trust-anchors anchors.pem --json options: --openssl PATH an OpenSSL 1.1+ binary (default: openssl on PATH) --profile PATH the producer's profile (PRF-01); its fields are checked and the export is checked against it `--trust-anchors` is a PEM file of the certificates the reader trusts for time stamping, taken from an EU Member State trusted list (a provider's service certificate or its issuing CA). An export that carries timestamp tokens is `refused` without it: nothing is evaluated. Licensed Apache-2.0. """ from __future__ import annotations import base64 import hashlib import json import os import re import shutil import subprocess import sys import tempfile from datetime import datetime, timezone RULEBOOK = "https://complyedge.io/rulebook/0.1.0" CHAIN_ALG = "sha256-chain-1" COMMITMENT_ALG = "hmac-sha256-1" MAX_SAFE_INT = 2 ** 53 # SPEC.md section 7: attributes that carry content and must never appear. CONTENT_ATTRIBUTES = { "gen_ai.input.messages", "gen_ai.output.messages", "gen_ai.system_instructions", "gen_ai.tool.call.arguments", "gen_ai.tool.call.result", "gen_ai.retrieval.documents", "gen_ai.retrieval.query.text", "gen_ai.memory.query.text", "gen_ai.memory.records", "gen_ai.evaluation.explanation", "gen_ai.prompt.variable", "gen_ai.guardrail.content.input.value", "gen_ai.guardrail.content.output.value", "message_context.prompt_text", "message_context.response_text", "input.value", "output.value", "message.content", "document.content", "text", "prompt", "completion", "content", } # Fields that hold a person (REC-08). Each must be a pseudonym object. PERSON_FIELDS = ("user", "session", "subject", "reviewer") EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$") PSEUDONYM_RE = re.compile(r"^p1:[0-9a-f]{32}$") SHA256_RE = re.compile(r"^[0-9a-f]{64}$") TS_RE = re.compile(r"^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}(\.[0-9]{3,9})?Z$") CODE_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:/-]*$") DECIMAL_01_RE = re.compile(r"^(0(\.[0-9]+)?|1(\.0+)?)$") ELI_TYPES = "reg|reg_del|reg_impl|dir|dir_del|dir_impl|dec|dec_del|dec_impl" ELI_OJ_RE = re.compile(r"^http://data\.europa\.eu/eli/(" + ELI_TYPES + r")/[0-9]{4}/[0-9]+" r"(/art_[0-9a-z]+)?(/par_[0-9a-z]+)?(/pnt_[0-9a-z]+)?(/anx_[ivxlc]+)?/oj$") ELI_NATIONAL_RE = re.compile(r"^https?://[^ ]+/eli/[^ ]+$") ELI_CONSOLIDATED_RE = re.compile(r"^http://data\.europa\.eu/eli/(" + ELI_TYPES + r")/[0-9]{4}/[0-9]+/[0-9]{4}-[0-9]{2}-[0-9]{2}$") EVENT_TYPES = {"decision", "oversight", "contest", "notice", "fallback", "logging_failure", "opt_out", "incident", "classification", "oversight_evaluation", "legal_hold"} HOLD_ACTIONS = {"placed", "lifted"} # REC-12: a locator is a URI (RFC 3986 scheme, then no whitespace), 255 characters at most. LOCATOR_RE = re.compile(r"^[A-Za-z][A-Za-z0-9+.-]*:\S+$") # REC-20: "at least six months" holds for any start date only from 184 days. SIX_MONTHS_DAYS = 184 PROFILE_REQUIRED = ["rulebook", "producer_id", "classes", "clock_source", "anchoring", "retention_classes", "copies", "event_types", "reason_codes", "rule_set_index_url", "interpretation_guide_url", "commitment_domain", "pseudonym_domain", "verification_interface_url"] VERDICTS = {"allow", "deny", "modify", "ask", "defer"} ACTIONS = {"passed", "blocked", "flagged", "modified", "escalated"} DIRECTIONS = {"input", "output", "action"} FALLBACK_KINDS = {"skipped", "degraded", "fallback_engine", "timeout"} FAILURE_KINDS = {"write_failed", "chain_failed", "anchor_failed"} OVERSIGHT_ACTIONS = {"accepted", "disregarded", "overridden", "reversed", "not_used", "intervened", "stopped", "verified"} OUTCOME_CHANGING = {"overridden", "reversed"} DECISION_MODES = {"solely_automated", "decisive", "advisory", "human"} ART22_EXCEPTIONS = {"22(2)(a)", "22(2)(b)", "22(2)(c)"} REQUEST_KINDS = {"human_intervention", "express_view", "explanation", "contest", "review"} REVIEW_OUTCOMES = {"upheld", "rectified", "withdrawn"} NOTICE_KINDS = {"use_of_ai", "rights", "explanation", "statement_of_reasons", "rejection"} INCIDENT_KINDS = {"risk", "substantial_modification", "serious_incident", "suspension"} IMPORTANCE = {"high", "medium", "low"} CLASSIFICATION_RESULTS = {"high_risk", "not_high_risk"} REQUIRED_RECORD = [ "rulebook", "record_version", "record_id", "occurred_at", "event_type", "account_id", "producer_id", "deploying_party", "retention", ] # REC-07: records about the account's chain, not one AI system, name no deciding system. CHAIN_LEVEL_FAILURES = {"anchor_failed", "chain_failed"} REQUIRED_BY_EVENT = { "decision": ["verdict", "action", "rules_fired", "rule_set", "exemptions", "commitments", "data_categories", "purpose_ref"], "oversight": ["proposal_ref", "oversight_action", "reviewer", "reviewer_role", "authority_ref", "review_opened_at", "inputs_ref", "reason_code"], "contest": ["decision_ref", "request_kind", "requested_at"], "notice": ["notice_kind", "decision_ref", "channel_ref", "language", "issued_at", "statement_ref"], "fallback": ["fallback"], "logging_failure": ["failure_kind", "detected_at"], "incident": ["incident_kind", "detected_at", "reported_at", "report_ref"], "classification": ["classification_result", "assessment_ref"], "oversight_evaluation": ["period", "action_counts", "evaluator_role", "representatives_informed"], "legal_hold": ["hold"], } # A decision record that carries `subject` is about a natural person (REC-08); # LAW-07, RGT-01, RGT-02 and RGT-04 then apply. REQUIRED_ABOUT_PERSON = ["legal_basis", "decision_mode", "notice", "explanation_inputs"] EXPLANATION_FIELDS = ["logic_ref", "procedure_ref", "main_factors", "data_categories", "risks_ref", "main_elements"] MAIN_ELEMENT_FIELDS = ["outcome_code", "effect_code", "effective_date"] # Fields that belong to an oversight record (OVS-02: never written on the decision). OVERSIGHT_ONLY = {"proposal_ref", "oversight_action", "reviewer", "review_opened_at", "outcome_ref"} ENVELOPE_REQUIRED = ["rulebook", "account_id", "producer_id", "period", "clock_source", "chain_summary", "anchors", "anchor_coverage", "retention_profile_url", "interpretation_guide_url"] INFORMATIONAL = {"unanchored_tail", "pre_chain", "no_anchor", "reference_unresolved"} NOT_CHECKED = {"anchor_token_not_checked"} TOKEN_FINDINGS = {"anchor_token_parse", "anchor_token_status", "anchor_token_imprint", "anchor_token_signature", "anchor_token_trust", "anchor_token_not_checked"} # --- canonical form (RFC 8785) ---------------------------------------------- def canonical(obj) -> bytes: """RFC 8785 for the value space a record may use: objects, arrays, strings, integers within +-(2^53 - 1), booleans and null. Non-integer numbers and larger integers are refused (REC-01).""" return _jcs(obj).encode("utf-8") def _jcs(v) -> str: if v is None: return "null" if v is True: return "true" if v is False: return "false" if isinstance(v, int): if abs(v) >= MAX_SAFE_INT: raise ValueError("integer outside +-(2^53 - 1); write it as a string (REC-01)") return str(v) if isinstance(v, float): raise ValueError("non-integer number; write it as a decimal string (REC-01)") if isinstance(v, str): return json.dumps(v, ensure_ascii=False) if isinstance(v, list): return "[" + ",".join(_jcs(x) for x in v) + "]" if isinstance(v, dict): if any(not isinstance(k, str) for k in v): raise ValueError("object keys must be strings") keys = sorted(v, key=lambda k: k.encode("utf-16-be")) # UTF-16 code units return "{" + ",".join(json.dumps(k, ensure_ascii=False) + ":" + _jcs(v[k]) for k in keys) + "}" raise ValueError(f"value of type {type(v).__name__} has no JSON form") def chain_hash(prev_hex: str, record_without_chain: dict) -> str: return hashlib.sha256(prev_hex.encode("utf-8") + canonical(record_without_chain)).hexdigest() def anchor_statement_bytes(statement: dict) -> bytes: return canonical(statement) def parse_ts(s): if not isinstance(s, str) or not TS_RE.match(s): return None frac = s[20:-1] base = datetime.strptime(s[:19], "%Y-%m-%dT%H:%M:%S").replace(tzinfo=timezone.utc) if frac: base = base.replace(microsecond=int(frac[:6].ljust(6, "0"))) return base # --- report ------------------------------------------------------------------- class Report: def __init__(self): self.findings: list[dict] = [] def add(self, finding: str, seq=None, record_id=None, detail: str = ""): self.findings.append({"finding": finding, "seq": seq, "record_id": record_id, "detail": detail}) def verdict(self) -> str: names = {f["finding"] for f in self.findings} if names - INFORMATIONAL - NOT_CHECKED: return "failed" if names: return "unverified" return "verified" def walk(obj, path=""): """Yield (path, key, value) for every member of every object, depth first.""" if isinstance(obj, dict): for k, v in obj.items(): p = path + "." + k if path else k yield p, k, v yield from walk(v, p) elif isinstance(obj, list): for i, v in enumerate(obj): yield from walk(v, f"{path}[{i}]") def _strings(obj, path=""): if isinstance(obj, str): yield path, obj elif isinstance(obj, dict): for k, v in obj.items(): yield from _strings(v, path + "." + k if path else k) elif isinstance(obj, list): for i, v in enumerate(obj): yield from _strings(v, f"{path}[{i}]") def check_content_and_numbers(obj, rep: Report, seq, rid, where: str): """REC-01 number form, REC-10 / section 7 content, REC-08 e-mail shapes, EXP-05.""" for path, key, val in walk(obj): if key in CONTENT_ATTRIBUTES or path in CONTENT_ATTRIBUTES: rep.add("content_present", seq, rid, f"{where}{path}") if isinstance(val, float): rep.add("invalid_value", seq, rid, f"{where}{path} is a non-integer number; REC-01 requires a string") elif isinstance(val, int) and not isinstance(val, bool) and abs(val) >= MAX_SAFE_INT: rep.add("invalid_value", seq, rid, f"{where}{path} is an integer outside +-(2^53 - 1); REC-01 requires a string") for path, val in _strings(obj): if EMAIL_RE.match(val.removeprefix("mailto:")): rep.add("person_in_clear", seq, rid, f"{where}{path} looks like an e-mail address") # --- records ----------------------------------------------------------------- def _pseudonym_ok(v) -> bool: return isinstance(v, dict) and bool(PSEUDONYM_RE.match(str(v.get("value", "")))) \ and isinstance(v.get("key_version"), int) and not isinstance(v.get("key_version"), bool) def _provision_problem(prov) -> str | None: """LAW-02: None when the provision object is well formed, else why not.""" if not isinstance(prov, dict): return "no provision object" if not prov.get("label"): return "provision lacks label" eli, url = prov.get("eli"), prov.get("url") if eli is None and url is None: return "provision lacks eli and url" if eli is not None and not (ELI_OJ_RE.match(str(eli)) or (ELI_NATIONAL_RE.match(str(eli)) and "data.europa.eu" not in str(eli))): return f"provision.eli is not in the Official Journal form: {eli}" if url is not None and not str(url).startswith("https://"): return "provision.url is not an https address" ce = prov.get("consolidated_eli") if ce is not None and not ELI_CONSOLIDATED_RE.match(str(ce)): return f"provision.consolidated_eli is not an act-level dated ELI: {ce}" return None def effective_event_type(rec: dict) -> str: et = rec.get("event_type") return et if et in EVENT_TYPES else "decision" # REC-05: unknown is read as decision def check_record(rec: dict, env: dict, rep: Report): seq = (rec.get("chain") or {}).get("seq") if isinstance(rec.get("chain"), dict) else None rid = rec.get("record_id") def missing(field, why=""): rep.add("required_field_missing", seq, rid, field + (f" ({why})" if why else "")) def invalid(detail): rep.add("invalid_value", seq, rid, detail) def need(obj, fields, prefix="", why=""): for f in fields: if not isinstance(obj, dict) or f not in obj: missing(prefix + f, why) def enum(field, value, allowed): if value is not None and value not in allowed: invalid(f"{field} is {value!r}, not one of {sorted(allowed)}") def ts_field(field, value): if value is not None and parse_ts(value) is None: invalid(f"{field} is not an RFC 3339 UTC timestamp") need(rec, REQUIRED_RECORD) if "rulebook" in rec and rec["rulebook"] != RULEBOOK: invalid(f"rulebook must be {RULEBOOK}") if "account_id" in rec and env.get("account_id") is not None and rec["account_id"] != env.get("account_id"): invalid("account_id differs from the envelope's account_id") for f in ("occurred_at", "detected_at", "review_opened_at", "requested_at", "responded_at", "rectified_at", "issued_at", "reported_at"): ts_field(f, rec.get(f)) et = effective_event_type(rec) need(rec, REQUIRED_BY_EVENT.get(et, []), why=f"required for event_type {et}") chain_level = et == "legal_hold" or (et == "logging_failure" and rec.get("failure_kind") in CHAIN_LEVEL_FAILURES) if not chain_level: need(rec, ["deciding_system"]) ds = rec.get("deciding_system") if isinstance(ds, dict): need(ds, ["id", "version"], "deciding_system.") ck = rec.get("checker") if isinstance(ck, dict) and ck.get("id") == ds.get("id") and ck.get("version") == ds.get("version"): invalid("deciding_system names the checker (REC-07)") ret = rec.get("retention") if isinstance(ret, dict): need(ret, ["class", "erase_after"], "retention.") if "legal_hold" in ret: invalid("retention.legal_hold: a hold is placed after the record is chained, so it is its own " "legal_hold record, never a field on the record it covers (REC-19)") ts_field("retention.erase_after", ret.get("erase_after")) check_content_and_numbers(rec, rep, seq, rid, "") # REC-08: every person is a pseudonym with a key version. for f in PERSON_FIELDS: if f in rec and not _pseudonym_ok(rec[f]): rep.add("person_in_clear", seq, rid, f) hd = rec.get("human_decider") if hd is not None: if not isinstance(hd, dict) or not _pseudonym_ok(hd.get("pseudonym")): rep.add("person_in_clear", seq, rid, "human_decider.pseudonym") need(hd, ["pseudonym", "role", "authority_ref"], "human_decider.") # REC-11 and CMT-01, CMT-04: commitments are keyed and carry a key version. directions = set() for i, c in enumerate(rec.get("commitments") or []): if not isinstance(c, dict): invalid(f"commitments[{i}] is not an object") continue need(c, ["alg", "value", "key_version", "direction"], f"commitments[{i}].") if "alg" in c and c["alg"] != COMMITMENT_ALG: invalid(f"commitments[{i}].alg is {c['alg']!r}: an unkeyed or unknown hash is not a commitment (CMT-01)") if "value" in c and not SHA256_RE.match(str(c["value"])): invalid(f"commitments[{i}].value is not 64 lower-case hex characters") enum(f"commitments[{i}].direction", c.get("direction"), DIRECTIONS) sr = c.get("source_ref") if sr is not None and not (isinstance(sr, str) and len(sr) <= 255 and LOCATOR_RE.match(sr)): invalid(f"commitments[{i}].source_ref is not a locator: a URI of at most 255 characters " "with no whitespace (REC-12)") directions.add(c.get("direction")) if {"input", "output"} <= directions and not rec.get("correlation_id"): missing("correlation_id", "required on a record about a request-output pair (REC-11)") # REC-13 dc = rec.get("data_categories") if dc is not None and not (isinstance(dc, list) and all(isinstance(x, str) for x in dc)): invalid("data_categories is not an array of labels") # REC-16: codes, never prose; confidence a decimal string in [0, 1]. for i, code in enumerate(rec.get("reason_codes") or []): if not isinstance(code, str) or not CODE_RE.match(code): invalid(f"reason_codes[{i}] is not a code (letters, digits, _ . : / -): free text is not allowed (REC-16)") for path, key, val in walk(rec): if key == "confidence" and not (isinstance(val, str) and DECIMAL_01_RE.match(val)): invalid(f"{path} must be a decimal string in [0, 1]") for f in ("reason_code",): if f in rec and not (isinstance(rec[f], str) and CODE_RE.match(rec[f])): invalid(f"{f} is not a code: free text is not allowed") if et == "decision": check_decision(rec, rep, seq, rid, missing, invalid, need, enum) elif et == "oversight": enum("oversight_action", rec.get("oversight_action"), OVERSIGHT_ACTIONS) if rec.get("oversight_action") in OUTCOME_CHANGING and "outcome_ref" not in rec: missing("outcome_ref", "required when the person changed the outcome (OVS-01)") elif et == "contest": enum("request_kind", rec.get("request_kind"), REQUEST_KINDS) enum("review_outcome", rec.get("review_outcome"), REVIEW_OUTCOMES) if "review_outcome" in rec: need(rec, ["responded_at", "reviewer", "statement_ref"], why="required once the request is answered (RGT-03)") if rec.get("review_outcome") == "rectified": need(rec, ["rectified_at"], why="required when the decision was rectified (RGT-03)") elif et == "notice": enum("notice_kind", rec.get("notice_kind"), NOTICE_KINDS) elif et == "fallback": check_fallback(rec.get("fallback"), need, enum) elif et == "logging_failure": enum("failure_kind", rec.get("failure_kind"), FAILURE_KINDS) elif et == "incident": enum("incident_kind", rec.get("incident_kind"), INCIDENT_KINDS) elif et == "legal_hold": hold = rec.get("hold") need(hold, ["id", "action", "scope"], "hold.") if isinstance(hold, dict): enum("hold.action", hold.get("action"), HOLD_ACTIONS) sc = hold.get("scope") need(sc, ["from_seq"], "hold.scope.") if isinstance(sc, dict): lo, hi = sc.get("from_seq"), sc.get("to_seq") if "from_seq" in sc and not (isinstance(lo, int) and not isinstance(lo, bool) and lo >= 1): invalid("hold.scope.from_seq is not a positive integer") elif hi is not None and not (isinstance(hi, int) and not isinstance(hi, bool) and hi >= lo): invalid("hold.scope.to_seq is not an integer at or after from_seq") elif et == "classification": enum("classification_result", rec.get("classification_result"), CLASSIFICATION_RESULTS) elif et == "oversight_evaluation": counts = rec.get("action_counts") if isinstance(counts, dict): for k, v in counts.items(): if k not in OVERSIGHT_ACTIONS: invalid(f"action_counts key {k!r} is not an oversight_action") elif not (isinstance(v, int) and not isinstance(v, bool) and v >= 0): invalid(f"action_counts.{k} is not a non-negative integer") if "reviewer_counts" in rec and env.get("export_purpose") != "oversight_evaluation": rep.add("reviewer_rates_exposed", seq, rid, "per-reviewer counts in an export whose purpose is not oversight evaluation (OVS-05)") if not isinstance(rec.get("representatives_informed", False), bool): invalid("representatives_informed is not a boolean") def check_fallback(fb, need, enum): if fb is None: return need(fb, ["kind", "reason_code"], "fallback.") if isinstance(fb, dict): enum("fallback.kind", fb.get("kind"), FALLBACK_KINDS) def check_decision(rec, rep, seq, rid, missing, invalid, need, enum): enum("verdict", rec.get("verdict"), VERDICTS) enum("action", rec.get("action"), ACTIONS) check_fallback(rec.get("fallback"), need, enum) ex = rec.get("exemptions") if isinstance(ex, dict): need(ex, ["claimed", "rejected"], "exemptions.") for kind in ("claimed", "rejected"): for i, e in enumerate(ex.get(kind) or []): need(e, ["name", "provision"], f"exemptions.{kind}[{i}].") if kind == "rejected": need(e, ["reason_code"], f"exemptions.{kind}[{i}].", "a rejected exemption names its reason (REC-17)") present = sorted(OVERSIGHT_ONLY & set(rec)) if present: invalid(f"oversight fields {present} on a decision record: the human action is its own record (OVS-02)") enum("decision_mode", rec.get("decision_mode"), DECISION_MODES) if rec.get("decision_mode") == "human" and "human_decider" not in rec: missing("human_decider", "required when decision_mode is human (OVS-04)") about_person = "subject" in rec if about_person: need(rec, REQUIRED_ABOUT_PERSON, why="required for a decision about a natural person") lb = rec.get("legal_basis") if lb is not None: need(lb, ["provision"], "legal_basis.") if isinstance(lb, dict): if "provision" in lb: why = _provision_problem(lb["provision"]) if why: rep.add("provision_unchecked", seq, rid, "legal_basis." + why) enum("legal_basis.art22_exception", lb.get("art22_exception"), ART22_EXCEPTIONS) nt = rec.get("notice") if nt is not None: need(nt, ["given"], "notice.") if isinstance(nt, dict): if not isinstance(nt.get("given", False), bool): invalid("notice.given is not a boolean") if nt.get("given") is True: need(nt, ["given_at", "notice_ref"], "notice.", "required once notice was given (RGT-02)") ts_ok = nt.get("given_at") is None or parse_ts(nt.get("given_at")) is not None if not ts_ok: invalid("notice.given_at is not an RFC 3339 UTC timestamp") if rec.get("subject_is_employee") is True: need(nt, ["workers_informed_at"], "notice.", "required for an employee (RGT-02, AI Act Art 26(7))") ei = rec.get("explanation_inputs") if ei is not None: need(ei, EXPLANATION_FIELDS, "explanation_inputs.", "RGT-04") if isinstance(ei, dict): for i, mf in enumerate(ei.get("main_factors") or []): if not isinstance(mf, dict): invalid(f"explanation_inputs.main_factors[{i}] is not an object") continue need(mf, ["code", "importance"], f"explanation_inputs.main_factors[{i}].") extra = sorted(set(mf) - {"code", "importance"}) if extra: invalid(f"explanation_inputs.main_factors[{i}] carries {extra}: factor codes and buckets only, never values (RGT-04)") enum(f"explanation_inputs.main_factors[{i}].importance", mf.get("importance"), IMPORTANCE) if isinstance(ei.get("main_elements"), dict): need(ei["main_elements"], MAIN_ELEMENT_FIELDS, "explanation_inputs.main_elements.") if rec.get("withheld") is True and "escrow_ref" not in rec: missing("escrow_ref", "required when information is withheld (RGT-05)") def check_law(rec: dict, rep: Report, rule_provisions: dict): if effective_event_type(rec) != "decision": return seq = (rec.get("chain") or {}).get("seq") if isinstance(rec.get("chain"), dict) else None rid = rec.get("record_id") rs = rec.get("rule_set") if "rule_set" in rec: rs_ok = isinstance(rs, dict) and SHA256_RE.match(str(rs.get("digest", ""))) and rs.get("label") \ and str(rs.get("index_url", "")).startswith("https://") if not rs_ok: rep.add("provision_unchecked", seq, rid, "rule_set lacks digest, label or index_url") for i, r in enumerate(rec.get("rules_fired") or []): if not isinstance(r, dict) or not r.get("rule_id") or not r.get("rule_version") or not r.get("citation"): rep.add("provision_unchecked", seq, rid, f"rules_fired[{i}] lacks rule_id, rule_version or citation") continue why = _provision_problem(r.get("provision")) if why: rep.add("provision_unchecked", seq, rid, f"rules_fired[{i}].{why}") if r.get("caller_asserted") is True: rep.add("provision_unchecked", seq, rid, f"rules_fired[{i}] is caller-asserted, presented as checked") prov = r.get("provision") if isinstance(r.get("provision"), dict) else {} target = None if why else (prov.get("eli") or prov.get("url")) seen = rule_provisions.setdefault(r["rule_id"], target) if target and seen and seen != target: rep.add("invalid_value", seq, rid, f"rule_id {r['rule_id']} names {target}; elsewhere in the export it names {seen} (LAW-01)") # --- references between records (OVS-01..03, RGT-03, RGT-07) ----------------- def check_references(records: list[dict], rep: Report): by_id = {} for r in records: rid = r.get("record_id") if rid is None: continue if rid in by_id: rep.add("invalid_value", None, rid, "record_id is not unique in the export (REC-03)") by_id.setdefault(rid, r) def seq_of(r): ch = r.get("chain") return ch.get("seq") if isinstance(ch, dict) else None def resolve(r, field, want_decision=True): ref = r.get(field) if ref is None: return None target = by_id.get(ref) if target is None: rep.add("reference_unresolved", seq_of(r), r.get("record_id"), f"{field} {ref} is not in this export; the pair cannot be checked") return None if want_decision and effective_event_type(target) != "decision": rep.add("invalid_value", seq_of(r), r.get("record_id"), f"{field} {ref} is not a decision record") return None if target is r: rep.add("invalid_value", seq_of(r), r.get("record_id"), f"{field} names the record itself (OVS-02)") return target for r in records: et = effective_event_type(r) if et == "oversight": resolve(r, "proposal_ref") resolve(r, "outcome_ref") elif et in ("contest", "notice"): resolve(r, "decision_ref") elif et == "decision" and "identification_ref" in r: ident = resolve(r, "identification_ref") if ident is None or r.get("action") == "passed": continue if not str(ident.get("annex_iii_point", "")).startswith("1(a)"): continue here = seq_of(r) verifiers = { (o.get("reviewer") or {}).get("value") for o in records if effective_event_type(o) == "oversight" and o.get("oversight_action") == "verified" and o.get("proposal_ref") == ident.get("record_id") and (here is None or seq_of(o) is None or seq_of(o) < here) } verifiers.discard(None) if len(verifiers) < 2: rep.add("dual_verification_missing", here, r.get("record_id"), f"acts on identification {ident.get('record_id')} with {len(verifiers)} distinct " "verifying reviewer(s) recorded before it; AI Act Art 14(5) needs two (OVS-03)") # --- envelope (EXP-01, EXP-05, CHN-05) ---------------------------------------- def check_envelope(env: dict, records: list[dict], rep: Report): for f in ENVELOPE_REQUIRED: if f not in env: rep.add("required_field_missing", None, None, "envelope." + f) if "rulebook" in env and env["rulebook"] != RULEBOOK: rep.add("invalid_value", None, None, f"envelope.rulebook must be {RULEBOOK}") cs = env.get("clock_source") if isinstance(cs, dict): for f in ("reference", "tolerance_ms"): if f not in cs: rep.add("required_field_missing", None, None, "envelope.clock_source." + f) for i, s in enumerate(env.get("signatures") or []): for f in ("alg", "key_ref", "value"): if not isinstance(s, dict) or f not in s: rep.add("required_field_missing", None, None, f"envelope.signatures[{i}].{f} (CHN-05)") check_content_and_numbers({k: v for k, v in env.items() if k != "anchors"}, rep, None, None, "envelope.") chained = [r for r in records if isinstance(r.get("chain"), dict)] summ = env.get("chain_summary") if isinstance(summ, dict) and chained is not None: seqs = [r["chain"].get("seq") for r in chained if isinstance(r["chain"].get("seq"), int)] actual = {"first_seq": min(seqs) if seqs else 0, "last_seq": max(seqs) if seqs else 0, "chained_rows": len(chained), "pre_chain_rows": len(records) - len(chained)} for k, v in actual.items(): if k in summ and summ[k] != v: rep.add("invalid_value", None, None, f"envelope.chain_summary.{k} is {summ[k]}; the records give {v}") # --- chain (CHN-01..04, EXP-02) ------------------------------------------------- def check_chain(records: list[dict], envelope: dict, rep: Report): tol_ms = (envelope.get("clock_source") or {}).get("tolerance_ms", 0) tol_ms = tol_ms if isinstance(tol_ms, int) else 0 chained = [r for r in records if isinstance(r.get("chain"), dict)] unchained = [r for r in records if not isinstance(r.get("chain"), dict)] for r in unchained: if r.get("chain_status") == "pre_chain": rep.add("pre_chain", None, r.get("record_id"), "labelled, not verified") else: rep.add("unchained_after_chain_start", None, r.get("record_id"), "no chain fields and not labelled pre_chain") if chained and unchained: times = [parse_ts(r.get("occurred_at")) for r in chained] times = [t for t in times if t] first = min(times) if times else None for r in unchained: t = parse_ts(r.get("occurred_at")) if first and r.get("chain_status") == "pre_chain" and t and t > first: rep.add("unchained_after_chain_start", None, r.get("record_id"), "labelled pre_chain but dated after the first chained record") chained = [r for r in chained if isinstance(r["chain"].get("seq"), int)] chained.sort(key=lambda r: r["chain"]["seq"]) seen = {} prev_hash, prev_seq, prev_ts = "", None, None for r in chained: ch = r["chain"] seq, rid = ch.get("seq"), r.get("record_id") for f in ("prev", "hash", "alg"): if f not in ch: rep.add("required_field_missing", seq, rid, "chain." + f) if "alg" in ch and ch["alg"] != CHAIN_ALG: rep.add("invalid_value", seq, rid, f"chain.alg is {ch['alg']!r}, not {CHAIN_ALG}") if seq in seen: rep.add("sequence_duplicate", seq, rid, f"also {seen[seq]}") continue seen[seq] = rid if prev_seq is None and seq != 1 and envelope.get("chain_summary", {}).get("first_seq") == 1: rep.add("sequence_gap", seq, rid, "the export starts after seq 1") if prev_seq is not None and seq != prev_seq + 1: rep.add("sequence_gap", seq, rid, f"after {prev_seq}") body = {k: v for k, v in r.items() if k != "chain"} try: expected = chain_hash(str(ch.get("prev", "")), body) except ValueError: expected = None # reported by the record check (REC-01) if expected is not None and ch.get("hash") != expected: rep.add("hash_mismatch", seq, rid, "stored hash is not the hash of the content and prev") if prev_seq is not None and seq == prev_seq + 1 and ch.get("prev") != prev_hash: rep.add("link_break", seq, rid, "prev is not the previous record's hash") if prev_seq is None and seq == 1 and ch.get("prev") not in ("", None): rep.add("link_break", seq, rid, "seq 1 must have an empty prev") t = parse_ts(r.get("occurred_at")) if t and prev_ts and (prev_ts - t).total_seconds() * 1000 > tol_ms: rep.add("backdated", seq, rid, "occurred_at earlier than the previous record beyond clock tolerance") prev_hash, prev_seq, prev_ts = ch.get("hash"), seq, t or prev_ts return chained # --- RFC 3161 tokens (ANC-02) --------------------------------------------------- # # Enough DER to read a TimeStampResp or a bare TimeStampToken: the PKIStatus, # the TSTInfo message imprint, genTime and accuracy, the embedded certificates # and the signer's serial number. The signature is checked by openssl. OID_SIGNED_DATA = bytes.fromhex("2a864886f70d010702") # 1.2.840.113549.1.7.2 OID_TST_INFO = bytes.fromhex("2a864886f70d0109100104") # 1.2.840.113549.1.9.16.1.4 OID_SHA256 = bytes.fromhex("608648016503040201") # 2.16.840.1.101.3.4.2.1 OID_EKU = bytes.fromhex("551d25") # 2.5.29.37 OID_TIME_STAMPING = bytes.fromhex("2b06010505070308") # 1.3.6.1.5.5.7.3.8 class DerError(ValueError): pass def _tlv(buf: bytes, pos: int): """One TLV at pos: (tag, value_start, value_end, tlv_end).""" if pos + 2 > len(buf): raise DerError("unexpected end of data") tag, length = buf[pos], buf[pos + 1] pos += 2 if length & 0x80: n = length & 0x7F if n == 0 or n > 4 or pos + n > len(buf): raise DerError("unsupported length encoding") length = int.from_bytes(buf[pos:pos + n], "big") pos += n if pos + length > len(buf): raise DerError("length exceeds data") return tag, pos, pos + length, pos + length def _children(buf: bytes, start: int, end: int): out, pos = [], start while pos < end: tag, vs, ve, pos = _tlv(buf, pos) out.append((tag, vs, ve)) return out def _gen_time(raw: bytes) -> datetime: s = raw.decode("ascii") if not s.endswith("Z"): raise DerError("time is not in UTC") s = s[:-1] frac = "" if "." in s: s, frac = s.split(".", 1) fmt = "%Y%m%d%H%M%S" if len(s) == 14 else "%y%m%d%H%M%S" t = datetime.strptime(s, fmt).replace(tzinfo=timezone.utc) if frac: t = t.replace(microsecond=int(frac[:6].ljust(6, "0"))) return t def parse_token(der: bytes) -> dict: """Parse a TimeStampResp or a TimeStampToken. Raises DerError if neither.""" tag, vs, ve, end = _tlv(der, 0) if tag != 0x30 or end != len(der): raise DerError("not a single DER SEQUENCE") top = _children(der, vs, ve) status, form, ci = 0, "token", (vs, ve) if top and top[0][0] == 0x30: # TimeStampResp: PKIStatusInfo first form = "response" st = _children(der, top[0][1], top[0][2]) if not st or st[0][0] != 0x02: raise DerError("PKIStatusInfo has no status") status = int.from_bytes(der[st[0][1]:st[0][2]], "big") if len(top) < 2: return {"form": form, "status": status} ci = (top[1][1], top[1][2]) token_der = der[_tlv_start(der, top[1])] else: token_der = der ci_parts = _children(der, *ci) if len(ci_parts) < 2 or ci_parts[0][0] != 0x06 or der[ci_parts[0][1]:ci_parts[0][2]] != OID_SIGNED_DATA: raise DerError("not a CMS SignedData") sd_seq = _children(der, ci_parts[1][1], ci_parts[1][2]) if not sd_seq or sd_seq[0][0] != 0x30: raise DerError("SignedData missing") sd = _children(der, sd_seq[0][1], sd_seq[0][2]) if len(sd) < 4 or sd[2][0] != 0x30: raise DerError("SignedData is incomplete") eci = _children(der, sd[2][1], sd[2][2]) if len(eci) < 2 or eci[0][0] != 0x06 or der[eci[0][1]:eci[0][2]] != OID_TST_INFO: raise DerError("content is not a TSTInfo") octets = _children(der, eci[1][1], eci[1][2]) if not octets or octets[0][0] != 0x04: raise DerError("TSTInfo octets missing") tst_buf = der[octets[0][1]:octets[0][2]] t_tag, t_vs, t_ve, _ = _tlv(tst_buf, 0) tst = _children(tst_buf, t_vs, t_ve) if len(tst) < 5 or tst[2][0] != 0x30 or tst[4][0] != 0x18: raise DerError("TSTInfo is incomplete") mi = _children(tst_buf, tst[2][1], tst[2][2]) alg = _children(tst_buf, mi[0][1], mi[0][2]) imprint_alg = tst_buf[alg[0][1]:alg[0][2]] imprint = tst_buf[mi[1][1]:mi[1][2]] gen_time = _gen_time(tst_buf[tst[4][1]:tst[4][2]]) accuracy_ms = None if len(tst) > 5 and tst[5][0] == 0x30: accuracy_ms = 0 for a_tag, a_vs, a_ve in _children(tst_buf, tst[5][1], tst[5][2]): n = int.from_bytes(tst_buf[a_vs:a_ve], "big") accuracy_ms += n * 1000 if a_tag == 0x02 else n if a_tag == 0x80 else n / 1000 if a_tag == 0x81 else 0 certs = [] signer_serial = None for tag_, cvs, cve in sd[3:]: if tag_ == 0xA0: # certificates [0] IMPLICIT for c_tag, c_vs, c_ve in _children(der, cvs, cve): start = _tlv_start(der, (c_tag, c_vs, c_ve)) certs.append(der[start]) elif tag_ == 0x31: # signerInfos sis = _children(der, cvs, cve) if sis: si = _children(der, sis[0][1], sis[0][2]) if len(si) > 1 and si[1][0] == 0x30: # issuerAndSerialNumber ias = _children(der, si[1][1], si[1][2]) signer_serial = der[ias[1][1]:ias[1][2]] return {"form": form, "status": status, "token_der": token_der, "imprint_alg": imprint_alg, "imprint": imprint, "gen_time": gen_time, "accuracy_ms": accuracy_ms, "certs": certs, "signer_serial": signer_serial} def _tlv_start(buf: bytes, child) -> slice: """Slice of the whole TLV whose value spans child[1]:child[2].""" _, vs, ve = child # Walk back over the length octets to the tag. for back in range(2, 7): start = vs - back if start < 0: break try: tag, s, e, end = _tlv(buf, start) except DerError: continue if s == vs and e == ve: return slice(start, ve) raise DerError("cannot locate TLV") def parse_cert(der: bytes) -> dict: tag, vs, ve, _ = _tlv(der, 0) tbs_tag, tvs, tve = _children(der, vs, ve)[0] tbs = _children(der, tvs, tve) i = 1 if tbs[0][0] == 0xA0 else 0 # optional [0] version serial = der[tbs[i][1]:tbs[i][2]] validity = _children(der, tbs[i + 3][1], tbs[i + 3][2]) not_before = _gen_time(der[validity[0][1]:validity[0][2]]) not_after = _gen_time(der[validity[1][1]:validity[1][2]]) eku = None for t, s, e in tbs[i + 6:]: if t == 0xA3: exts = _children(der, *_children(der, s, e)[0][1:]) for _, xs, xe in exts: parts = _children(der, xs, xe) if der[parts[0][1]:parts[0][2]] == OID_EKU: octet = parts[-1] inner = der[octet[1]:octet[2]] _, ivs, ive, _ = _tlv(inner, 0) eku = {inner[s2:e2] for _, s2, e2 in _children(inner, ivs, ive)} return {"serial": serial, "not_before": not_before, "not_after": not_after, "eku": eku} def pem_to_der_list(text: str) -> list[bytes]: out = [] for block in re.findall(r"-----BEGIN CERTIFICATE-----(.*?)-----END CERTIFICATE-----", text, re.S): out.append(base64.b64decode("".join(block.split()))) return out def der_to_pem(ders: list[bytes]) -> bytes: out = [] for d in ders: b = base64.encodebytes(d).decode().replace("\n", "") out.append("-----BEGIN CERTIFICATE-----\n" + "\n".join(b[i:i + 64] for i in range(0, len(b), 64)) + "\n-----END CERTIFICATE-----\n") return "".join(out).encode() def find_openssl(preferred: str | None = None) -> str | None: """An OpenSSL 1.1+ binary (`ts -verify -partial_chain`), or None. LibreSSL, the openssl macOS ships, lacks the option and is reported as unusable.""" path = preferred or shutil.which("openssl") if not path: return None try: out = subprocess.run([path, "version"], capture_output=True, text=True, timeout=10).stdout except (OSError, subprocess.SubprocessError): return None return path if out.startswith("OpenSSL") and "LibreSSL" not in out else None def openssl_verify(openssl: str, token: dict, statement: bytes, ca_pem: bytes) -> tuple[str, str]: """('ok', '') | ('failed', reason) | ('not_checked', reason).""" with tempfile.TemporaryDirectory() as tmp: paths = {n: os.path.join(tmp, n) for n in ("token", "data", "ca")} with open(paths["token"], "wb") as fh: fh.write(token["token_der"]) with open(paths["data"], "wb") as fh: fh.write(statement) with open(paths["ca"], "wb") as fh: fh.write(ca_pem) cmd = [openssl, "ts", "-verify", "-token_in", "-in", paths["token"], "-data", paths["data"], "-CAfile", paths["ca"], "-partial_chain", "-attime", str(int(token["gen_time"].timestamp()))] try: proc = subprocess.run(cmd, capture_output=True, text=True, timeout=30) except (OSError, subprocess.SubprocessError) as exc: return "not_checked", f"openssl did not run: {exc}" out = (proc.stdout + proc.stderr).strip() if proc.returncode == 0 and "Verification: OK" in out: return "ok", "" if "Verification: FAILED" in out: errs = [ln for ln in out.splitlines() if "error" in ln.lower()] return "failed", (errs[-1] if errs else out.splitlines()[-1])[-200:] return "not_checked", f"openssl could not check the token: {out[-200:]}" def check_token(anchor: dict, statement: bytes, trust_pem: bytes, openssl: str | None) -> list[tuple[str, str]]: """Findings for one anchor's token, in step order; stops at the first failed step.""" try: der = base64.b64decode(anchor.get("token_b64") or "", validate=True) tok = parse_token(der) except (ValueError, IndexError) as exc: return [("anchor_token_parse", f"not an RFC 3161 TimeStampResp or TimeStampToken: {exc}")] if tok["status"] not in (0, 1): return [("anchor_token_status", f"PKIStatus {tok['status']}: the provider did not grant the timestamp")] if "token_der" not in tok: return [("anchor_token_parse", "granted response carries no token")] signer = None for c in tok["certs"]: try: pc = parse_cert(c) except (ValueError, IndexError): continue if pc["serial"] == tok["signer_serial"]: signer = (c, pc) if signer is not None: pc = signer[1] if not (pc["not_before"] <= tok["gen_time"] <= pc["not_after"]): return [("anchor_token_status", f"genTime {tok['gen_time']:%Y-%m-%dT%H:%M:%SZ} is outside the signing " f"certificate's validity ({pc['not_before']:%Y-%m-%d} to {pc['not_after']:%Y-%m-%d})")] if pc["eku"] is None or OID_TIME_STAMPING not in pc["eku"]: return [("anchor_token_status", "the signing certificate is not for time stamping (no timeStamping purpose)")] if tok["accuracy_ms"] is not None and tok["accuracy_ms"] > 1000: return [("anchor_token_status", f"declared accuracy {tok['accuracy_ms']} ms is worse than one second (ANC-02)")] if tok["imprint_alg"] != OID_SHA256: return [("anchor_token_imprint", "message imprint is not SHA-256 (ANC-01)")] if tok["imprint"] != hashlib.sha256(statement).digest(): return [("anchor_token_imprint", "the token is not over this anchor statement")] if openssl is None: return [("anchor_token_not_checked", "signature and trust need OpenSSL 1.1 or later (LibreSSL lacks -partial_chain)")] if signer is not None: outcome, why = openssl_verify(openssl, tok, statement, der_to_pem([signer[0]])) if outcome == "failed": return [("anchor_token_signature", why)] if outcome == "not_checked": return [("anchor_token_not_checked", why)] outcome, why = openssl_verify(openssl, tok, statement, trust_pem) if outcome == "failed": step = "anchor_token_trust" if signer is not None else "anchor_token_signature" return [(step, why if signer is None else "the signing certificate does not chain to a pinned trust anchor: " + why)] if outcome == "not_checked": return [("anchor_token_not_checked", why)] return [] def check_anchors(chained: list[dict], envelope: dict, rep: Report, trust_pem, openssl): anchors = envelope.get("anchors") or [] if not chained: return by_seq = {r["chain"]["seq"]: r for r in chained} last_seq = max(by_seq) if not anchors: rep.add("no_anchor", None, None, "the export carries no anchor") return newest = 0 for i, a in enumerate(anchors): st = a.get("statement") if isinstance(a, dict) else None if not isinstance(st, dict): rep.add("required_field_missing", None, None, f"envelope.anchors[{i}].statement") continue for f in ("token_b64", "provider_url", "stored_at"): if f not in a: rep.add("required_field_missing", None, None, f"envelope.anchors[{i}].{f}") extra = sorted(set(st) - {"account_id", "chain_alg", "chain_hash", "chain_seq"}) if extra: rep.add("content_present", None, None, f"envelope.anchors[{i}].statement carries {extra} (ANC-01)") s = st.get("chain_seq") if not isinstance(s, int): rep.add("anchor_mismatch", None, None, f"anchors[{i}] has no chain_seq") continue if s > last_seq: rep.add("truncated_tail", s, None, f"anchor attests seq {s}; export ends at {last_seq}") continue r = by_seq.get(s) if r is None: rep.add("anchor_mismatch", s, None, "no record at the anchored seq") continue if st.get("chain_hash") != r["chain"]["hash"] or st.get("account_id") != r.get("account_id") \ or st.get("chain_alg") != CHAIN_ALG: rep.add("anchor_mismatch", s, r.get("record_id"), "anchor statement does not match the export at that seq") continue for name, why in check_token(a, anchor_statement_bytes(st), trust_pem, openssl): rep.add(name, s, r.get("record_id"), why) newest = max(newest, s) # EXP-01: the coverage the envelope states must be the coverage its anchors claim. claimed = max((a["statement"]["chain_seq"] for a in anchors if isinstance(a, dict) and isinstance(a.get("statement"), dict) and isinstance(a["statement"].get("chain_seq"), int) and a["statement"]["chain_seq"] <= last_seq), default=0) cov = envelope.get("anchor_coverage") if isinstance(cov, dict): if cov.get("newest_anchored_seq") not in (None, claimed): rep.add("invalid_value", None, None, f"envelope.anchor_coverage.newest_anchored_seq is {cov.get('newest_anchored_seq')}; the anchors give {claimed}") if cov.get("unanchored_rows") not in (None, max(0, last_seq - claimed)): rep.add("invalid_value", None, None, f"envelope.anchor_coverage.unanchored_rows is {cov.get('unanchored_rows')}; the anchors give {max(0, last_seq - claimed)}") if newest and newest < last_seq: rep.add("unanchored_tail", newest + 1, None, f"records {newest + 1}..{last_seq} are newer than the newest anchor") # --- producer profile (PRF-01, REC-19, REC-20, CMT-02, CMT-05, CMT-06) --------- def check_profile(profile, env: dict, records: list[dict], rep: Report): """The profile's own fields, then the export against what the profile says. What the profile states about the producer's processes stays attested; this checks that it is stated and that the export agrees with it.""" if not isinstance(profile, dict): rep.add("required_field_missing", None, None, "profile is not a JSON object") return def bad(detail): rep.add("invalid_value", None, None, "profile." + detail) for f in PROFILE_REQUIRED: if f not in profile: rep.add("required_field_missing", None, None, "profile." + f) if "rulebook" in profile and profile["rulebook"] != RULEBOOK: bad(f"rulebook must be {RULEBOOK}") check_content_and_numbers(profile, rep, None, None, "profile.") classes = {} for i, rc in enumerate(profile.get("retention_classes") or []): if not isinstance(rc, dict) or not {"class", "period_days", "high_risk"} <= set(rc): rep.add("required_field_missing", None, None, f"profile.retention_classes[{i}] needs class, period_days, high_risk (REC-20)") continue days = rc["period_days"] if not (isinstance(days, int) and not isinstance(days, bool) and days >= 1): bad(f"retention_classes[{i}].period_days is not a positive integer") elif rc["high_risk"] is True and days < SIX_MONTHS_DAYS: bad(f"retention_classes[{i}] keeps high-risk records {days} days, under six months (REC-20)") classes[rc["class"]] = rc for i, cp in enumerate(profile.get("copies") or []): if not isinstance(cp, dict) or not {"what", "gone_after_days"} <= set(cp): rep.add("required_field_missing", None, None, f"profile.copies[{i}] needs what and gone_after_days (CMT-05)") cd, pd = profile.get("commitment_domain"), profile.get("pseudonym_domain") if isinstance(cd, str) and isinstance(pd, str) and (not cd or not pd or cd.startswith(pd) or pd.startswith(cd)): bad("commitment_domain and pseudonym_domain must be non-empty and neither a prefix of the other (CMT-02)") for f in ("rule_set_index_url", "interpretation_guide_url", "verification_interface_url"): if f in profile and not str(profile[f]).startswith("https://"): bad(f"{f} is not an https address") def mismatch(seq, rid, detail): rep.add("profile_mismatch", seq, rid, detail) if "producer_id" in profile and "producer_id" in env and profile["producer_id"] != env["producer_id"]: mismatch(None, None, "envelope.producer_id is not the profile's producer_id") if "clock_source" in profile and "clock_source" in env and profile["clock_source"] != env["clock_source"]: mismatch(None, None, "envelope.clock_source differs from the profile's clock_source") events = set(profile.get("event_types") or []) codes = set(profile.get("reason_codes") or []) for r in records: seq = (r.get("chain") or {}).get("seq") if isinstance(r.get("chain"), dict) else None rid = r.get("record_id") cls = (r.get("retention") or {}).get("class") if isinstance(r.get("retention"), dict) else None if "retention_classes" in profile and cls is not None and cls not in classes: mismatch(seq, rid, f"retention.class {cls!r} is not a class the profile defines (REC-19)") if "event_types" in profile and r.get("event_type") not in events: mismatch(seq, rid, f"event_type {r.get('event_type')!r} is not listed in the profile (PRF-01)") if "reason_codes" in profile: used = set(r.get("reason_codes") or []) for path, key, val in walk(r): if key == "reason_code" and isinstance(val, str): used.add(val) for code in sorted(used - codes): mismatch(seq, rid, f"reason code {code!r} is not listed in the profile (PRF-01)") # --- entry points -------------------------------------------------------------- def verify(export: dict, trust_anchors: bytes | None = None, openssl: str | None | bool = True, profile: dict | None = None) -> dict: """Verify one export. trust_anchors: PEM bytes of the pinned time-stamping trust anchors, or None. openssl: True to find one on PATH, a path, or None/False to run without it. profile: the producer's profile (PRF-01), or None to skip the profile checks. """ env = export.get("envelope") if isinstance(export.get("envelope"), dict) else {} records = export.get("records") if isinstance(export.get("records"), list) else [] has_tokens = any(isinstance(a, dict) and a.get("token_b64") for a in env.get("anchors") or []) if has_tokens and not trust_anchors: return {"rulebook": RULEBOOK, "verdict": "refused", "findings": [], "records": len(records), "chained": 0, "refused_reason": "the export carries timestamp tokens and no trust anchors were supplied"} tool = find_openssl(None if openssl is True else openssl) if openssl else None rep = Report() if not isinstance(export.get("envelope"), dict) or not isinstance(export.get("records"), list): rep.add("required_field_missing", None, None, "envelope and records") check_envelope(env, records, rep) rule_provisions: dict = {} for r in records: if not isinstance(r, dict): rep.add("invalid_value", None, None, "a record is not a JSON object") continue check_record(r, env, rep) check_law(r, rep, rule_provisions) records = [r for r in records if isinstance(r, dict)] check_references(records, rep) chained = check_chain(records, env, rep) check_anchors(chained, env, rep, trust_anchors, tool) if profile is not None: check_profile(profile, env, records, rep) return {"rulebook": RULEBOOK, "verdict": rep.verdict(), "findings": rep.findings, "records": len(records), "chained": len(chained)} def main(argv): args = argv[1:] if not args or args[0].startswith("-"): print(__doc__) return 2 trust, openssl = None, True if "--trust-anchors" in args: with open(args[args.index("--trust-anchors") + 1], "rb") as fh: trust = fh.read() if "--openssl" in args: openssl = args[args.index("--openssl") + 1] profile = None if "--profile" in args: with open(args[args.index("--profile") + 1], encoding="utf-8") as fh: profile = json.load(fh) with open(args[0], encoding="utf-8") as fh: export = json.load(fh) result = verify(export, trust, openssl, profile) if "--json" in args: print(json.dumps(result, indent=2, default=str)) else: print(f"verdict: {result['verdict']} records: {result['records']} chained: {result['chained']}") if result.get("refused_reason"): print(" " + result["refused_reason"]) for f in result["findings"]: print(f" {f['finding']:<28} seq={f['seq']} record_id={f['record_id']} {f['detail']}") return {"verified": 0, "unverified": 0, "failed": 1, "refused": 3}[result["verdict"]] if __name__ == "__main__": sys.exit(main(sys.argv))