"""XCIM C01 offline consent verifier. No email-message inputs or network I/O. The supplied test anchor adapter is NOT a blockchain verifier. Its results cannot establish CONSENT_CURRENT or XCIM_PASS. See README.txt and test-report.json. """ from __future__ import annotations import base64 import hashlib import hmac import json import re from copy import deepcopy from pathlib import Path from urllib.parse import urlsplit import idna from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey from jsonschema import Draft202012Validator PROFILE = "xcim-consent-01" TYP = "application/xcim-consent+json" SAFE = 9007199254740991 ROOT = Path(__file__).resolve().parent ROLES = {"sender": "sender_binding", "consent": "consent_attestation", "revocation": "consent_revocation", "status": "consent_status", "anchor_batch": "anchor_publication"} HASH = {"type": "string", "pattern": "^[A-Za-z0-9_-]{43}$"} URI = {"type": "string", "minLength": 1, "maxLength": 512, "pattern": r"^https://[!-~]+$"} TEXT = {"type": "string", "minLength": 1, "maxLength": 4096} ANCHOR_ID = {"type": "string", "minLength": 1, "maxLength": 512, "pattern": r"^[\x20-\x7e]+$"} TOKEN = {"type": "string", "minLength": 1, "maxLength": 64, "pattern": r"^[A-Za-z0-9._-]+$"} UINT = {"type": "integer", "minimum": 0, "maximum": SAFE} POSINT = {**UINT, "minimum": 1} UUID = {"type": "string", "pattern": r"^urn:uuid:[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$"} class C01Error(ValueError): def __init__(self, code, detail=""): self.code = code super().__init__(code + (": " + detail if detail else "")) def require(condition, code, detail=""): if not condition: raise C01Error(code, detail) def closed(properties): return {"type": "object", "properties": properties, "required": list(properties), "additionalProperties": False} def array(items, minimum=1, maximum=64): return {"type": "array", "items": items, "minItems": minimum, "maxItems": maximum, "uniqueItems": True} def payload(kind, properties): return closed({"profile": {"const": PROFILE}, "kind": {"const": kind}, "issuer": URI, **properties}) def schemas(): defs = {} defs["jws"] = closed({ "protected": {"type": "string", "pattern": "^[A-Za-z0-9_-]+$"}, "payload": {"type": "string", "pattern": "^[A-Za-z0-9_-]+$"}, "signature": {"type": "string", "pattern": "^[A-Za-z0-9_-]{86}$"}, }) defs["protected"] = closed({"alg": {"const": "Ed25519"}, "kid": {"type": "string", "minLength": 1, "maxLength": 200, "pattern": r"^[\x20-\x7e]+$"}, "typ": {"const": TYP}}) defs["sender"] = payload("sender", { "sender_id": URI, "sender_epoch": POSINT, "name": TEXT, "scope": TEXT, "entity_evidence_commitment": HASH, "issued_at": UINT, "expires_at": UINT}) defs["consent"] = payload("consent", { "consent_id": UUID, "sender_id": URI, "sender_epoch": POSINT, "sender_binding_commitment": HASH, "address_profile": {"const": "ascii-idna15-1"}, "recipient_salt": HASH, "recipient_commitment": HASH, "types": array(closed({"type_id": URI, "version": POSINT, "title": TEXT, "description": TEXT})), "notice": closed({"language": {"type": "string", "minLength": 1, "maxLength": 35, "pattern": r"^[\x20-\x7e]+$"}, "text": TEXT}), "control": closed({"method": {"const": "oidc-01"}, "policy_id": URI, "observed_at": UINT, "evidence_commitment": HASH}), "decision": {"const": "explicit_consent"}, "accepted_at": UINT, "issued_at": UINT, "expires_at": UINT}) defs["revocation"] = payload("revocation", { "event_id": UUID, "consent_id": UUID, "consent_commitment": HASH, "sequence": {**POSINT, "maximum": 64}, "previous_event_commitment": HASH, "type_ids": array(URI), "actor": {"enum": ["recipient", "sender", "attester"]}, "method": TOKEN, "evidence_commitment": HASH, "requested_at": UINT, "effective_at": UINT, "issued_at": UINT}) defs["status"] = payload("status", { "consent_id": UUID, "consent_commitment": HASH, "sender_id": URI, "sender_epoch": POSINT, "sequence": {**UINT, "maximum": 64}, "event_head": HASH, "active_type_ids": array(URI, 0), "this_update": UINT, "next_update": UINT}) defs["anchor_batch"] = payload("anchor_batch", { "log_id": URI, "sequence": POSINT, "previous_batch_commitment": {"oneOf": [HASH, {"type": "null"}]}, "created_at": UINT, "leaves": array(HASH, 1, 4096)}) # Anchor fields are structurally present; exact type/encoding belongs to the # pre-admitted chain profile. No core assumption of 32-byte chain block IDs. defs["anchor_evidence"] = closed({"profile_id": URI, "network_id": ANCHOR_ID, "genesis_hash": ANCHOR_ID, "anchor_id": ANCHOR_ID, "block_hash": ANCHOR_ID, "transaction_hash": ANCHOR_ID, "record_index": UINT, "batch_commitment": HASH, "proof": {"type": "object"}}) defs["anchor"] = closed({"batch": {"$ref": "#/$defs/jws"}, "evidence": {"$ref": "#/$defs/anchor_evidence"}}) defs["evidence"] = closed({"profile": {"const": PROFILE}, "sender": {"$ref": "#/$defs/jws"}, "consent": {"$ref": "#/$defs/jws"}, "revocations": array({"$ref": "#/$defs/jws"}, 0), "status": {"$ref": "#/$defs/jws"}, "consent_anchor": {"$ref": "#/$defs/anchor"}, "revocation_anchors": array(closed({"event_id": UUID, "batch": {"$ref": "#/$defs/jws"}, "evidence": {"$ref": "#/$defs/anchor_evidence"}}), 0)}) return {"$schema": "https://json-schema.org/draft/2020-12/schema", "$id": "https://xcim.org/schemas/c01/consent.schema.json", "$comment": "Local draft candidate. Structural validation is insufficient; see CONTRACT.txt.", "$defs": defs, "$ref": "#/$defs/evidence"} SCHEMA = schemas() VALIDATORS = {kind: Draft202012Validator({"$defs": SCHEMA["$defs"], "$ref": "#/$defs/" + kind}) for kind in SCHEMA["$defs"]} def b64(data): return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii") def unb64(value, length=None): require(isinstance(value, str) and re.fullmatch(r"[A-Za-z0-9_-]+", value), "BASE64URL") try: decoded = base64.b64decode(value + "=" * (-len(value) % 4), altchars=b"-_", validate=True) except (ValueError, base64.binascii.Error) as exc: raise C01Error("BASE64URL") from exc require(b64(decoded) == value and (length is None or len(decoded) == length), "BASE64URL") return decoded def jcs(value): """RFC 8785 restricted to the contract's nonnegative safe-integer domain.""" def encode(node): if node is None: return "null" if isinstance(node, bool): return "true" if node else "false" if isinstance(node, str): node.encode("utf-8", errors="strict") return json.dumps(node, ensure_ascii=False, separators=(",", ":")) if type(node) is int: require(0 <= node <= SAFE, "JSON_INTEGER") return str(node) if isinstance(node, list): return "[" + ",".join(encode(x) for x in node) + "]" if isinstance(node, dict): require(all(isinstance(k, str) for k in node), "JSON_INVALID") return "{" + ",".join(encode(k) + ":" + encode(node[k]) for k in sorted(node, key=lambda s: s.encode("utf-16-be"))) + "}" raise C01Error("JSON_INTEGER") try: return encode(value).encode("utf-8") except UnicodeError as exc: raise C01Error("JSON_UNICODE") from exc def strict_json(raw, limit=2 * 1024 * 1024, canonical=False): require(isinstance(raw, bytes) and len(raw) <= limit, "JSON_LIMIT") try: text = raw.decode("utf-8", errors="strict") depth, quoted, escaped = 0, False, False for c in text: if quoted: if escaped: escaped = False elif c == "\\": escaped = True elif c == '"': quoted = False elif c == '"': quoted = True elif c in "[{": depth += 1 require(depth <= 16, "JSON_DEPTH") elif c in "]}": depth -= 1 def pairs(items): result = {} for k, v in items: require(k not in result, "JSON_DUPLICATE") result[k] = v return result def integer(token): require(not token.startswith("-") and len(token) <= 16, "JSON_INTEGER") result = int(token) require(result <= SAFE, "JSON_INTEGER") return result def forbidden(_): raise C01Error("JSON_INTEGER") value = json.loads(text, object_pairs_hook=pairs, parse_int=integer, parse_float=forbidden, parse_constant=forbidden) encoded = jcs(value) require(not canonical or encoded == raw, "JSON_NONCANONICAL") return value except C01Error: raise except (ValueError, UnicodeError, RecursionError) as exc: raise C01Error("JSON_INVALID") from exc def valid_uri(value): try: value.encode("ascii") parsed = urlsplit(value) parsed.port # Raises on invalid port text or out-of-range ports. return (len(value) <= 512 and parsed.scheme == "https" and bool(parsed.hostname) and parsed.username is None and parsed.password is None and not parsed.query and not parsed.fragment and "?" not in value and "#" not in value and "\\" not in value and not re.search(r"%(?![0-9a-fA-F]{2})", value) and bool(re.fullmatch(r"[A-Za-z0-9:/\[\]@!$&'()*+,;=._~%\-]+", value))) except (ValueError, UnicodeError): return False def validate(kind, value): errors = list(VALIDATORS[kind].iter_errors(value)) require(not errors, "SCHEMA", str(errors[0].message) if errors else "") # JSON Schema patterns alone do not implement canonical base64 or URI policy. def walk(node): if isinstance(node, dict): for k, v in node.items(): if k in {"issuer", "sender_id", "type_id", "policy_id", "log_id", "profile_id"}: require(valid_uri(v), "IDENTIFIER_URI") if k in {"type_ids", "active_type_ids"}: require(all(valid_uri(x) for x in v), "IDENTIFIER_URI") if k.endswith("_commitment") or k in {"recipient_salt", "event_head"}: if v is not None: unb64(v, 32) if k != "proof": walk(v) elif isinstance(node, list): for item in node: walk(item) walk(value) def commitment(value): return b64(hashlib.sha256(b"XCIM-C01-OBJECT\0" + jcs(value)).digest()) def canonical_address(value): require(isinstance(value, str) and idna.idnadata.__version__ == "15.1.0", "ADDRESS_PROFILE") value = value.strip(" \t") require(value.count("@") == 1 and not any(c.isspace() for c in value), "ADDRESS_SYNTAX") local, domain = value.split("@") require(re.fullmatch(r"[A-Za-z0-9!#$%&'*+/=?^_`{|}~.-]+", local) and 1 <= len(local) <= 64 and not local.startswith(".") and not local.endswith(".") and ".." not in local and not domain.endswith("."), "ADDRESS_SYNTAX") try: domain = idna.encode(domain, uts46=True, transitional=False, std3_rules=True).decode("ascii").lower() except idna.IDNAError as exc: raise C01Error("ADDRESS_SYNTAX") from exc require(len(domain) <= 253 and all(1 <= len(s) <= 63 for s in domain.split(".")) and len(local + "@" + domain) <= 254, "ADDRESS_SYNTAX") return local + "@" + domain def recipient_commitment(consent, address): values = [consent[k] for k in ("issuer", "consent_id", "sender_id", "sender_epoch", "address_profile", "recipient_salt")] + [canonical_address(address)] return b64(hashlib.sha256(b"XCIM-C01-RECIPIENT\0" + jcs(values)).digest()) def sign(value, key, kid): """Fixture utility. Production issuer transaction and key custody are external.""" protected = b64(jcs({"alg": "Ed25519", "kid": kid, "typ": TYP})) data = b64(jcs(value)) return {"protected": protected, "payload": data, "signature": b64(key.sign((protected + "." + data).encode("ascii")))} def verify_signed(envelope, kind, policy): require(len(jcs(envelope)) <= (512 if kind == "anchor_batch" else 128) * 1024, "ENVELOPE_LIMIT") validate("jws", envelope) header = strict_json(unb64(envelope["protected"]), 128 * 1024, True) validate("protected", header) obj = strict_json(unb64(envelope["payload"]), 256 * 1024 if kind == "anchor_batch" else 128 * 1024, True) if isinstance(obj, dict) and obj.get("profile") != PROFILE: raise C01Error("UNSUPPORTED_PROFILE") validate(kind, obj) require(obj["issuer"] == policy["issuer"], "ISSUER") key = policy["keys"].get(header["kid"]) require(key is not None and ROLES[kind] in key["roles"] and not key.get("revoked", False) and key.get("historical_accepted", True), "KEY_ROLE") issued_at = obj.get("issued_at", obj.get("this_update", obj.get("created_at"))) require(key["not_before"] <= issued_at < key["not_after"], "KEY_TIME") try: Ed25519PublicKey.from_public_bytes(unb64(key["public_key"], 32)).verify( unb64(envelope["signature"], 64), (envelope["protected"] + "." + envelope["payload"]).encode("ascii")) except (InvalidSignature, ValueError) as exc: raise C01Error("SIGNATURE") from exc return obj def sorted_unique(values): return values == sorted(set(values)) class FloorStore: """In-memory demonstration only. Deployments MUST persist authenticated floors. The caller must restore a trustworthy state or decline current acceptance. This example does not implement disk durability, administrator recovery, or HA. """ def __init__(self): self.records = {} self.type_definitions = {} self.type_conflicts = set() def observe_definitions(self, grant, policy_known): definitions = [] for item in grant["types"]: key = jcs([grant["issuer"], grant["sender_id"], grant["sender_epoch"], item["type_id"], item["version"]]).decode("utf-8") definition = {"title": item["title"], "description": item["description"]} definitions.append((key, definition)) if key in self.type_conflicts: return False for known in (self.type_definitions.get(key), policy_known.get(key)): if known is not None and known != definition: self.type_conflicts.add(key) return False for key, definition in definitions: self.type_definitions[key] = definition return True def adopt(self, other): self.records = deepcopy(other.records) self.type_definitions = deepcopy(other.type_definitions) self.type_conflicts = set(other.type_conflicts) def observe(self, consent_hash, sequence, head, active, removed, history): candidate = {"sequence": sequence, "head": head, "active": sorted(active), "removed": sorted(removed), "history": list(history), "conflicted": False} old = self.records.get(consent_hash) if old: if old["conflicted"]: old["removed"] = sorted(set(old["removed"]) | removed) return "CONFLICTED_HISTORY", old common = min(old["sequence"], sequence) if old["history"][:common] != list(history)[:common]: old["conflicted"] = True old["removed"] = sorted(set(old["removed"]) | removed) return "CONFLICTED_HISTORY", old if old["sequence"] > sequence: return "OLDER_THAN_KNOWN_FLOOR", old if old["sequence"] == sequence and (old["head"] != head or old["active"] != sorted(active)): old["conflicted"] = True old["removed"] = sorted(set(old["removed"]) | removed) return "CONFLICTED_HISTORY", old if not set(old["removed"]).issubset(removed): old["conflicted"] = True old["removed"] = sorted(set(old["removed"]) | removed) return "CONFLICTED_HISTORY", old self.records[consent_hash] = candidate return "ACCEPTED_FLOOR", candidate def verify_anchor(anchor, obj, policy, adapters): batch = verify_signed(anchor["batch"], "anchor_batch", policy) require(sorted_unique(batch["leaves"]), "BATCH_ORDER") for leaf in batch["leaves"]: unb64(leaf, 32) require((batch["sequence"] == 1) == (batch["previous_batch_commitment"] is None), "BATCH_PREVIOUS") require(batch["log_id"] in policy["log_ids"], "LOG_AUTHORITY") require(commitment(obj) in batch["leaves"] and batch["created_at"] >= obj["issued_at"], "ANCHOR_MEMBER") digest = commitment(batch) wrapper = anchor["evidence"] require(wrapper["batch_commitment"] == digest, "ANCHOR_COMMITMENT") admitted = policy["anchor_profiles"].get(wrapper["profile_id"]) adapter = adapters.get(wrapper["profile_id"]) if admitted is None or adapter is None: return "UNSUPPORTED_PROFILE" result = adapter(deepcopy(wrapper), deepcopy(admitted)) require(isinstance(result, dict) and set(result) == {"status", "batch_commitment", "context"}, "ADAPTER_RESULT") if result["status"] in {"VERIFIED", "VERIFIED_TEST_ONLY"}: require(result["batch_commitment"] == digest and result["context"] == admitted["context"], "ADAPTER_BINDING") require(result["status"] in {"VERIFIED", "VERIFIED_TEST_ONLY", "UNAVAILABLE", "INVALID"}, "ADAPTER_RESULT") return result["status"] def fixture_anchor_adapter(wrapper, admitted): """Never verifies a chain. Equality checks only; cannot establish CURRENT.""" context = {k: wrapper[k] for k in ("network_id", "genesis_hash", "anchor_id")} if wrapper["proof"] == {"test_only": "unavailable"}: return {"status": "UNAVAILABLE", "batch_commitment": None, "context": context} if context != admitted["context"] or wrapper["proof"] != {"test_only": "no-consensus-proof"}: return {"status": "INVALID", "batch_commitment": None, "context": context} return {"status": "VERIFIED_TEST_ONLY", "batch_commitment": wrapper["batch_commitment"], "context": context} def verify(evidence_bytes, query, policy, adapters, floors=None): """Verify consent without accepting any email, body, header, or message ID. policy, query and adapter implementation are authenticated local inputs, never trusted because an evidence document supplies them. The harness's can_assert_xcim_pass is always false, even for restrictive results. """ result = {"outcome": "INVALID_EVIDENCE", "reason": "UNVALIDATED", "can_assert_xcim_pass": False, "consent_checks_result": None, "anchor_result": None, "pending_revocation_anchors": []} def finish(outcome, reason): result.update(outcome=outcome, reason=reason) return result try: ev = strict_json(evidence_bytes) if isinstance(ev, dict) and ev.get("profile") != PROFILE: return finish("UNSUPPORTED_PROFILE", "PROFILE") validate("evidence", ev) # Local context is not transport evidence and must contain exactly these # fields. In particular an email cannot become a hidden verification input. require(set(query) == {"canonical_recipient", "issuer", "sender_id", "sender_epoch", "type_id", "type_version", "evaluation_time"}, "QUERY_FIELDS") require(type(query["evaluation_time"]) is int and 0 <= query["evaluation_time"] <= SAFE, "QUERY_TIME") t, u = query["evaluation_time"], policy["clock_uncertainty"] require(type(u) is int and 0 <= u <= 60, "CLOCK_POLICY") if not policy["not_before"] <= t - u or t + u >= policy["expires_at"]: return finish("INDETERMINATE", "POLICY_TIME") require(policy.get("third_party_admitted") is True, "ISSUER_NOT_ADMITTED") sender = verify_signed(ev["sender"], "sender", policy) grant = verify_signed(ev["consent"], "consent", policy) events = [verify_signed(x, "revocation", policy) for x in ev["revocations"]] status = verify_signed(ev["status"], "status", policy) status_kid = strict_json(unb64(ev["status"]["protected"]), canonical=True)["kid"] status_key = policy["keys"][status_kid] if not status_key["not_before"] <= t - u or t + u >= status_key["not_after"]: return finish("INDETERMINATE", "STATUS_KEY_TIME") gh = commitment(grant) require(sender["issued_at"] < sender["expires_at"], "SENDER_TIME") require(grant["sender_binding_commitment"] == commitment(sender) and all(grant[k] == sender[k] for k in ("sender_id", "sender_epoch")), "SENDER_BINDING") require(sender["sender_epoch"] >= policy["sender_epoch_floors"].get(sender["sender_id"], 1), "SENDER_EPOCH") require(grant["control"]["policy_id"] in policy["control_policies"], "CONTROL_POLICY") require(grant["accepted_at"] - grant["control"]["observed_at"] <= policy["control_policies"][grant["control"]["policy_id"]]["max_observation_age"], "CONTROL_AGE") require(grant["control"]["observed_at"] <= grant["accepted_at"] <= grant["issued_at"] < grant["expires_at"] <= sender["expires_at"] and sender["issued_at"] <= grant["accepted_at"], "CONSENT_TIME") tids = [x["type_id"] for x in grant["types"]] require(sorted_unique(tids), "TYPES_ORDER") active, removed, head = set(tids), set(), gh last_effective = last_issued = grant["issued_at"] event_ids = set() for seq, event in enumerate(events, 1): require(event["event_id"] not in event_ids, "EVENT_ID") event_ids.add(event["event_id"]) require(event["consent_id"] == grant["consent_id"] and event["consent_commitment"] == gh and event["sequence"] == seq and event["previous_event_commitment"] == head, "EVENT_CHAIN") require(sorted_unique(event["type_ids"]) and set(event["type_ids"]).issubset(active), "EVENT_TYPES") require(event["method"] in policy["revocation_methods"].get(event["actor"], []), "ACTOR_METHOD") require(grant["issued_at"] <= event["requested_at"] <= event["effective_at"] <= event["issued_at"] and last_effective <= event["effective_at"] and last_issued <= event["issued_at"], "EVENT_TIME") require(event["issued_at"] <= t - u, "EVENT_FUTURE") removed.update(event["type_ids"]) active.difference_update(event["type_ids"]) head = commitment(event) last_effective, last_issued = event["effective_at"], event["issued_at"] require(all(status[k] == grant[k] for k in ("consent_id", "sender_id", "sender_epoch")) and status["consent_commitment"] == gh, "STATUS_SCOPE") require(status["sequence"] == len(events) and status["event_head"] == head and status["active_type_ids"] == sorted(active), "STATUS_PROJECTION") require(last_issued <= status["this_update"] < status["next_update"] <= min(grant["expires_at"], sender["expires_at"]) and status["next_update"] - status["this_update"] <= 300, "STATUS_INTERVAL") anchor_ids = [x["event_id"] for x in ev["revocation_anchors"]] require(len(anchor_ids) == len(set(anchor_ids)) and set(anchor_ids).issubset(event_ids), "REVOCATION_ANCHOR_SCOPE") # Tuple mismatches are not evidence corruption. No matching permission is # inferred from a different company, address, epoch, type or definition. recipient = canonical_address(query["canonical_recipient"]) if (query["issuer"] != grant["issuer"] or query["sender_id"] != grant["sender_id"] or query["sender_epoch"] != grant["sender_epoch"] or not any(x["type_id"] == query["type_id"] and x["version"] == query["type_version"] for x in grant["types"]) or not hmac.compare_digest(recipient_commitment(grant, recipient), grant["recipient_commitment"])): return finish("NO_MATCH", "QUERY_TUPLE") # Exact type/version has immutable meaning within an issuer/sender epoch. # These definitions are learned only after signed scope and projection # validation, and may also be pinned by authenticated local policy. definition_store = deepcopy(floors) if floors is not None else FloorStore() if not definition_store.observe_definitions(grant, policy.get("type_definitions", {})): if floors is not None: floors.type_conflicts = set(definition_store.type_conflicts) return finish("INDETERMINATE", "TYPE_DEFINITION_CONFLICT") # A malformed or invalid signature/scope/projection never mutates floors. # Authenticated terminal restrictions need no finalized chain to apply. floor_outcome, record = "ACCEPTED_FLOOR", {"removed": sorted(removed)} staged_floors = None if floors is not None: staged_floors = definition_store floor_outcome, record = staged_floors.observe(gh, len(events), head, active, removed, [commitment(x) for x in events]) # Retain authenticated restrictive events and signed history conflicts # independently of their anchor. A new active-only floor remains staged # until required proof structure has been checked below. if events or floor_outcome != "ACCEPTED_FLOOR": floors.adopt(staged_floors) if query["type_id"] in set(record["removed"]) | removed: # Presence of an anchor wrapper is not evidence that it was verified. # Withdrawal applies now; classify any ancillary anchors separately. pending = set(event_ids) for item in ev["revocation_anchors"]: event = next(x for x in events if x["event_id"] == item["event_id"]) try: if verify_anchor(item, event, policy, adapters) == "VERIFIED": pending.discard(item["event_id"]) except C01Error: pass # Invalid/missing chain evidence cannot restore a withdrawal. result["pending_revocation_anchors"] = sorted(pending) return finish("CONSENT_REVOKED", "AUTHENTICATED_WITHDRAWAL") if floor_outcome != "ACCEPTED_FLOOR": return finish("INDETERMINATE", floor_outcome) if t - u >= min(grant["expires_at"], sender["expires_at"]): return finish("CONSENT_EXPIRED", "FINITE_VALIDITY") # Future/stale status may not be used to extend evidence validity. if not (status["this_update"] <= t - u and t + u < min(status["next_update"], grant["expires_at"], sender["expires_at"], policy["expires_at"], status_key["not_after"]) and t + u - status["this_update"] <= 300): return finish("INDETERMINATE", "STATUS_FRESHNESS") for item in ev["revocation_anchors"]: event = next(x for x in events if x["event_id"] == item["event_id"]) ar = verify_anchor(item, event, policy, adapters) if ar != "VERIFIED": result["pending_revocation_anchors"].append(item["event_id"]) result["pending_revocation_anchors"] += sorted(event_ids - set(anchor_ids)) ar = verify_anchor(ev["consent_anchor"], grant, policy, adapters) result["anchor_result"] = ar result["consent_checks_result"] = "CONSENT_CURRENT" if ar == "UNSUPPORTED_PROFILE": return finish("UNSUPPORTED_PROFILE", "ANCHOR_PROFILE") if ar == "INVALID": return finish("INVALID_EVIDENCE", "ANCHOR_PROOF") if staged_floors is not None: floors.adopt(staged_floors) if ar != "VERIFIED": return finish("INDETERMINATE", "ANCHOR_TEST_ONLY" if ar == "VERIFIED_TEST_ONLY" else "ANCHOR_UNAVAILABLE") return finish("CONSENT_CURRENT", "ALL_MANDATORY_CHECKS") except C01Error as exc: return finish("UNSUPPORTED_PROFILE" if exc.code == "UNSUPPORTED_PROFILE" else "INVALID_EVIDENCE", exc.code) if __name__ == "__main__": (ROOT / "schemas").mkdir(exist_ok=True) (ROOT / "schemas" / "consent.schema.json").write_text(json.dumps(SCHEMA, indent=2) + "\n", encoding="utf-8") print("Generated closed C01 schema; no network or blockchain verification performed.")