"""Reference validator for the Lending Deal-Package Readiness standard (v0.5). Four checks: 1. STRUCTURE — the pack validates against deal-package.schema.json (jsonschema is a HARD dependency of the reference validator: without it, validation FAILS loudly rather than passing silently). 2. CATALOGUE — every field exists in the catalogue; provenance kind is COMPATIBLE with the field's catalogue tag; verified sources match the catalogue's named source; readiness.missing entries reference real fields. 3. READINESS — semantic rules (the standard's teeth): oven-ready ⇒ missing == [] and entity gate passed gaps-remaining ⇒ missing non-empty L2/L3 ⇒ verify-tagged fields present must not be self-declared (not_collected = honest absence), ≥1 verified fact exists L3 ⇒ attested + oven-ready attested ⇒ non-empty truth_statement AND an attestation object 4. ENTITY GATE — oven-ready / L2+ / attested packs require a corporate form (ltd/llp/plc). sole_trader / partnership lending can fall inside the UK regulated-credit perimeter (RAO art 60B, subject to the art 60C exemptions) — v0 does not cover regulated agreements, so such packs cannot claim readiness. 5. REQUIRED — requirement profiles: the pack's `the_ask.facility_type` selects a per-product profile from requirement-profiles.v0.json (invoice finance requires the debtor book; mezzanine the capital stack; …); packs without a facility_type — or with a taxonomy-only one — fall back to the catalogue's baseline `required_fields`. Every required field must be PROVIDED (value not None) or listed in readiness.missing — you cannot hide a required gap. 6. INSTANCES — repeatable sections (catalogue repeatable:true — people, guarantor, security_assets) may appear as '
:2', '
:3', … with per-fact provenance per instance; instances of non-repeatable sections are rejected; suffixes are canonical (no leading zeros); an instance requires the bare first instance to be present; readiness.missing may only cite instances that exist in the pack. 7. DATATYPES — every non-null value is checked against the catalogue's datatype (monetary/integer/decimal numeric, boolean, ISO date, list_string, url/string/text/document_ref string) and ratio is a decimal fraction in [0,1] — 0.7, never 70 (coverage multiples such as DSCR use datatype decimal). `check_artifacts(catalogue)` validates the standard's own artifacts against each other (field shape, vocabulary, code-list/source references, profile coverage) — the repeatable form of the adversarial audit's consistency battery. `validate_disposition(msg, catalogue)` validates a funder DISPOSITION message (disposition.schema.json): structure, status/decline codes against the code lists, and info_requested.field_ids against the catalogue. `stale_facts(pack, as_of, max_age_days)` is a separate freshness HELPER (verified facts older than N days) — kept OUT of validate() so validation stays clock-free. Run `python validate.py` for the self-check against examples/example-pack.ready.json. """ from __future__ import annotations import json import math import re from decimal import ROUND_HALF_UP, Decimal from functools import lru_cache from pathlib import Path _HERE = Path(__file__).parent _CORPORATE_FORMS = {"ltd", "llp", "plc"} # The versions this validator + bundled catalogue implement. A pack declaring a # different major.minor is REJECTED rather than silently validated under these # artifacts (version-confusion / replay-under-changed-catalogue — sec review IR3). _SUPPORTED_SCHEMA_MAJMIN = ("0", "5") _SUPPORTED_CATALOGUE_MAJMIN = ("0", "11") _SUPPORTED_DISPOSITION_MAJMIN = ("0", "2") # disposition.schema.json _SUPPORTED_CERTIFICATE_MAJMIN = ("0", "1") # borrowing-base-certificate.schema.json _MAX_DEPTH = 64 # bounded-input guard (sec review F6/IR6): reject pathologically nested documents _MAX_NODES = 200_000 # breadth guard (re-review #6): reject documents with too many members _MAX_STR = 100_000 # per-string cap (re-review-3 #9): one 1GB string bypasses node/depth caps _MAX_ERRORS = 1_000 # error-list cap (re-review-3 #11): a wide invalid pack can't return unbounded _MAX_STAGE_ERRORS = 200 # per-stage cap so a flood of field errors can't bury a gate refusal (V-07) # NB: a byte-size limit belongs at the transport/HTTP boundary BEFORE parsing (see spec # §9a Security considerations) — these in-validator caps bound an already-parsed object. def _load(rel: str) -> dict: return json.loads((_HERE / rel).read_text(encoding="utf-8")) def _safe(s) -> str: """repr() an attacker-controlled id/value for error output — escapes control characters (log/terminal-injection defence, sec review F4/IR11) and bounds length.""" r = repr(str(s)) return r if len(r) <= 120 else r[:117] + "…" # bare ellipsis: no assumed quote style (L11) def _reject_nonfinite(tok): raise ValueError(f"non-finite number '{tok}' is not valid JSON (NaN/Infinity rejected)") def _no_duplicate_keys(pairs): seen: dict = {} for k, v in pairs: if k in seen: # bound + repr the key (re-review-3 #21): a duplicate key is attacker-controlled raise ValueError(f"duplicate object key {_safe(k)} — ambiguous across implementations") seen[k] = v return seen def load_document(text: str) -> dict: """RECOMMENDED strict ingestion for UNTRUSTED pack/disposition bytes (sec review F2/F9/IR12): UTF-8 JSON that REJECTS non-finite numbers (NaN/Infinity) and duplicate object keys, so a Python consumer cannot bless bytes a conforming Go/Java/JS parser would reject or read differently. Implementers SHOULD parse via this, not bare json.loads. Deeply-nested input raises ValueError, not RecursionError (re-review-2 L1) — the ingestion path for untrusted bytes must not crash the caller on the input class it exists to reject. Also rejects `1e999`-style overflow-to-infinity (parse_constant only sees the literal NaN/Infinity tokens — re-review-3 C1) and lone surrogates (C2), both of which other languages reject or read differently and which would later crash canonical_json.""" try: doc = json.loads(text, parse_constant=_reject_nonfinite, parse_float=_finite_float, object_pairs_hook=_no_duplicate_keys) except RecursionError: raise ValueError("document nesting is too deep to parse safely") if _has_lone_surrogate(doc): raise ValueError("document contains a lone UTF-16 surrogate (not interoperable JSON)") return doc def _finite_float(tok: str) -> float: f = float(tok) if not math.isfinite(f): # '1e999' -> inf without ever hitting parse_constant (C1) raise ValueError(f"non-finite number '{tok}' is not valid JSON (overflow to Infinity)") return f def _first_non_nfc(obj, _d: int = 0) -> str | None: """The first string (value OR key) not in Unicode NFC form, else None. RFC 8785 deliberately does not normalise, so NFC and NFD forms of visually identical text canonicalize to different bytes — different pack_ref, different evidence_hash. macOS hands back NFD from its filesystem, so two producers of the "same" pack can disagree on its identity. We do NOT normalise on the producer's behalf (that would silently rewrite their data, and would break RFC 8785 conformance in `evidence.jcs`); we tell them, so they can emit NFC.""" import unicodedata if _d > _MAX_DEPTH: return None if isinstance(obj, str): return obj if unicodedata.normalize("NFC", obj) != obj else None if isinstance(obj, dict): for k, v in obj.items(): if isinstance(k, str) and unicodedata.normalize("NFC", k) != k: return k found = _first_non_nfc(v, _d + 1) if found is not None: return found return None if isinstance(obj, list): for v in obj: found = _first_non_nfc(v, _d + 1) if found is not None: return found return None def _has_lone_surrogate(obj, _d: int = 0) -> bool: """True if any string (value or key) contains a lone UTF-16 surrogate (C2).""" if _d > _MAX_DEPTH: return False if isinstance(obj, str): return any(0xD800 <= ord(c) <= 0xDFFF for c in obj) if isinstance(obj, dict): return any((isinstance(k, str) and _has_lone_surrogate(k, _d + 1)) or _has_lone_surrogate(v, _d + 1) for k, v in obj.items()) if isinstance(obj, list): return any(_has_lone_surrogate(v, _d + 1) for v in obj) return False def _too_deep(obj, limit: int = _MAX_DEPTH, _d: int = 0) -> bool: if _d > limit: return True if isinstance(obj, dict): return any(_too_deep(v, limit, _d + 1) for v in obj.values()) if isinstance(obj, list): return any(_too_deep(v, limit, _d + 1) for v in obj) return False def _too_broad(obj, limit: int = _MAX_NODES) -> bool: """True if the object graph has more than `limit` container members (re-review #6): the shape schema permits arbitrary section/field names, so an attacker could send millions of fields. Depth-bounded traversal; short-circuits once the cap is exceeded.""" stack = [(obj, 0)] count = 0 while stack: cur, d = stack.pop() if d > _MAX_DEPTH: continue # depth handled separately; don't recurse past the cap here if isinstance(cur, dict): count += len(cur) if count > limit: return True stack.extend((v, d + 1) for v in cur.values()) elif isinstance(cur, list): count += len(cur) if count > limit: return True stack.extend((v, d + 1) for v in cur) return False def _oversized_string(obj, _d: int = 0) -> bool: """True if any string in the graph exceeds _MAX_STR chars (re-review-3 #9): the node and depth caps count members/nesting, so a single gigabyte string slips past both.""" if _d > _MAX_DEPTH: return False if isinstance(obj, str): return len(obj) > _MAX_STR if isinstance(obj, dict): return any((isinstance(k, str) and len(k) > _MAX_STR) or _oversized_string(v, _d + 1) for k, v in obj.items()) if isinstance(obj, list): return any(_oversized_string(v, _d + 1) for v in obj) return False def _has_nonfinite(obj, _d: int = 0) -> bool: """True if any float in the object graph is NaN/±Infinity (sec review F2 + re-review #8). Depth-bounded so it cannot itself recurse away on hostile input; a document over the depth cap is refused separately before this runs.""" if _d > _MAX_DEPTH: return False if isinstance(obj, bool): return False if isinstance(obj, float): return not math.isfinite(obj) if isinstance(obj, dict): # check keys too (re-review-2 L6): a programmatically built {float('nan'): 1} # would otherwise slip past and crash later in canonical_json (allow_nan=False). return (any(isinstance(k, float) and not math.isfinite(k) for k in obj) or any(_has_nonfinite(v, _d + 1) for v in obj.values())) if isinstance(obj, list): return any(_has_nonfinite(v, _d + 1) for v in obj) return False def artifact_errors(required: tuple[str, ...] | None = None) -> list[str]: """Fail-LOUD on the validator's own trust-root artifacts (sec review F1/IR7). A missing or unparseable REQUIRED artifact must make the validator refuse the pack, not silently disable enum/profile/overlay gates. `required` lets a caller scope the set — e.g. disposition validation needs only the code lists, not the requirement profiles (re-review-2 L9).""" errs: list[str] = [] # `is None` not truthiness (re-review-3 L1): required=() must mean "check nothing", # not fall through to the full default set. if required is None: required = ("code-lists.v0.json", "requirement-profiles.v0.json", "reference-data.v0.json") for rel in required: try: json.loads((_HERE / rel).read_text(encoding="utf-8")) except (FileNotFoundError, IsADirectoryError, PermissionError): errs.append(f"DEGRADED: required artifact {rel} is missing — the validator cannot " "enforce code-list/profile/registry rules; refusing to bless the pack") except ValueError as e: errs.append(f"DEGRADED: required artifact {rel} is unparseable ({e}) — " "refusing to bless the pack") return errs def _catalogue_index(catalogue: dict) -> dict[str, dict]: """'section.field' -> catalogue field row. Defensive over a malformed catalogue (re-review-2 M3): skip entries missing id/fields rather than KeyError — check_artifacts reports the malformation, and validate() refuses on it before the semantic gates run.""" idx: dict[str, dict] = {} for s in catalogue.get("sections") or []: sid = s.get("id") if isinstance(s, dict) else None if not sid: continue for fld in (s.get("fields") or []): fid = fld.get("id") if isinstance(fld, dict) else None if fid: idx[f"{sid}.{fid}"] = fld return idx def _split_sid(sid: str) -> tuple[str, str | None]: """'guarantor:2' -> ('guarantor', '2'); 'guarantor' -> ('guarantor', None).""" base, sep, inst = str(sid).partition(":") return base, (inst if sep else None) def _norm_key(key: str) -> str: """Catalogue lookup key for a possibly instance-qualified 'sec[:n].field'.""" sid, _, fid = str(key).partition(".") return f"{_split_sid(sid)[0]}.{fid}" def _instance_ref_error(ref: str, sections_by_id: dict) -> str | None: """Validate the instance grammar of a 'sec[:n].field' reference (re-review #10): a canonical suffix (`[2-9]|[1-9][0-9]+`, no leading zero, no Unicode digits) on a section the catalogue marks repeatable. Returns an error string or None. Used where there is no pack context to prove the instance exists (disposition field_ids).""" sid = str(ref).partition(".")[0] base, inst = _split_sid(sid) if inst is None: return None sec = sections_by_id.get(base) if sec is None: return f"references unknown section: {_safe(ref)}" if not sec.get("repeatable"): return f"references an instance of non-repeatable section '{base}': {_safe(ref)}" if not re.fullmatch(r"[2-9]|[1-9][0-9]+", inst): return f"bad instance suffix {_safe(sid)} (canonical: '{base}:2', '{base}:3', …)" return None # ── 1. structure ──────────────────────────────────────────────────────────── def validate_structure(pack: dict) -> list[str]: """Validate against the JSON Schema. jsonschema is REQUIRED — the reference validator refuses to bless a pack it could not fully check.""" try: import jsonschema # type: ignore except ImportError: return ["DEGRADED: the 'jsonschema' library is not installed — structure " "was NOT validated. Install jsonschema; a pack cannot pass without it."] schema = _load("deal-package.schema.json") validator = jsonschema.Draft202012Validator(schema) return [f"{'/'.join(map(str, e.path))}: {e.message}" for e in validator.iter_errors(pack)] # ── 2. catalogue + tag/provenance compatibility ───────────────────────────── _TAG_KINDS = { "verify": {"verified", "not_collected"}, "declare": {"self-declared", "not_collected"}, "doc": {"document", "not_collected"}, } _DATATYPE_VOCAB = {"string", "text", "monetary_amount", "integer", "decimal", "ratio", "date", "boolean", "enum", "document_ref", "list_string", "url"} # Section/field ids are interpolated into generated patternProperties regexes and # into ':instance' parsing — constrain them to safe lowercase tokens (sec review F10). _ID_RE = re.compile(r"^[a-z][a-z0-9_]*$") def _D(v) -> Decimal: """Decimal from a JSON number WITHOUT inheriting binary float error. `Decimal(0.1)` is 0.1000000000000000055511151231257827; `Decimal(repr(0.1))` is exactly 0.1. Going via repr() (which is the shortest round-tripping decimal) is what makes money arithmetic behave the way an accountant expects. """ return Decimal(v) if isinstance(v, int) and not isinstance(v, bool) else Decimal(repr(v)) def _q(d: Decimal, places: int = 2) -> Decimal: """Quantize to the penny, rounding HALF UP — the convention in UK financial reporting (Python's default is banker's rounding, which is not what a certificate expects).""" return d.quantize(Decimal(1).scaleb(-places), rounding=ROUND_HALF_UP) def _money(v, places: int = 2) -> Decimal | None: """A JSON number as quantized Decimal, or None if it is not a usable number.""" return _q(_D(v), places) if _is_finite_number(v) else None def _is_finite_number(val) -> bool: """A real JSON number: int or float, not bool, not NaN/±Infinity (sec review F2/IR-NaN).""" if isinstance(val, bool): return False if isinstance(val, int): return True return isinstance(val, float) and math.isfinite(val) def _norm_sic(code) -> str | None: """Canonical ASCII digits for a SIC code, or None when the value cannot be read as one (fresh-eyes review V-01). A prefix match on a raw string is only as good as its normalization: `re.sub(r"[\\s-]")` misses ZERO WIDTH SPACE, ZWNJ, SOFT HYPHEN and the bidi marks — all invisible, none matched by `\\s`, and each enough to make "92000" (an excluded activity) miss the "92" prefix while still rendering as the excluded code to a human. Strip the whole format/control class, then REQUIRE plain ASCII digits so anything else fails closed.""" if not isinstance(code, str): return None import unicodedata cleaned = "".join(ch for ch in code if not (ch.isspace() or ch in "-‐‑‒–—―" # hyphens/dashes or unicodedata.category(ch) in ("Cf", "Cc"))) # invisible return cleaned if cleaned.isascii() and cleaned.isdigit() else None def _is_blank(val) -> bool: """True if a value is present-but-substantively-empty (re-review-2 H1): None, an empty/whitespace-only string, or an empty list/dict. A required field holding "" or [] or " " must NOT count as provided, and such a value is not a verified FACT.""" if val is None: return True if isinstance(val, str): return not val.strip() if isinstance(val, (list, dict)): return len(val) == 0 return False # numbers (incl. 0) and booleans (incl. False) are substantive def _iso_datetime_error(val) -> str | None: """None if val is an ISO date (YYYY-MM-DD) or RFC-3339-style date-time; error text otherwise (re-review #14/#15, re-review-2 M7). Rejects the space-separated and week-date forms datetime.fromisoformat would otherwise accept (parser-differential).""" if not isinstance(val, str) or not val: return "expected an ISO date or date-time" import datetime if re.fullmatch(r"[0-9]{4}-[0-9]{2}-[0-9]{2}", val): try: datetime.date.fromisoformat(val) return None except ValueError: return "not a valid calendar date" # date-time: require the 'T' separator + real parse (no space form, no week dates) if not re.match(r"^[0-9]{4}-[0-9]{2}-[0-9]{2}T", val): return "expected an ISO date (YYYY-MM-DD) or date-time (YYYY-MM-DDThh:mm:ss)" s = val[:-1] + "+00:00" if val.endswith("Z") else val # tolerate trailing Z try: datetime.datetime.fromisoformat(s) return None except ValueError: return "not a valid ISO date-time" def _datatype_error(datatype: str, val) -> str | None: """Check a non-None value against its catalogue datatype; error text or None.""" if datatype in ("string", "text", "document_ref", "enum", "url"): return None if isinstance(val, str) else f"expected a string ({datatype})" if datatype == "date": # Anchored, real-calendar check (sec review F8): a loose ^\d{4}-\d{2}-\d{2} # prefix-match passed "2026-13-45trailing" and Unicode digits. Require exactly # YYYY-MM-DD ASCII digits AND a date that actually exists. if not isinstance(val, str) or not re.fullmatch(r"[0-9]{4}-[0-9]{2}-[0-9]{2}", val): return "expected an ISO date (YYYY-MM-DD)" import datetime try: datetime.date.fromisoformat(val) except ValueError: return "not a valid calendar date (YYYY-MM-DD)" return None if datatype == "boolean": return None if isinstance(val, bool) else "expected a boolean" if datatype == "integer": # Align with JSON Schema's mathematical integer (sec review IR5): a strict # schema's {"type":"integer"} accepts 1.0, so the reference validator must too, # or a pack passes one layer and fails the other. Reject bool and non-finite. if isinstance(val, bool): return "expected an integer" if isinstance(val, int): return None if isinstance(val, float) and math.isfinite(val) and val.is_integer(): return None return "expected an integer" if datatype in ("monetary_amount", "decimal"): return None if _is_finite_number(val) else f"expected a number ({datatype})" if datatype == "ratio": if not _is_finite_number(val): return "expected a number (ratio)" return None if 0 <= val <= 1 else \ "ratio is a decimal fraction in [0,1] (0.7, never 70; coverage multiples use decimal)" if datatype == "list_string": return None if isinstance(val, list) and all(isinstance(x, str) for x in val) \ else "expected an array of strings (list_string)" return None # unknown datatype: check_artifacts flags the catalogue, not the pack def _dict_key(rel: str, key: str) -> dict: """Load a top-level dict `key` from artifact `rel`, tolerating a malformed file (re-review-3 #6): a non-object file or non-object value yields {} rather than an AttributeError deep in a gate. Genuine absence/corruption is surfaced by artifact_errors() / check_artifacts(); this only keeps the loaders from crashing.""" try: doc = _load(rel) except (OSError, ValueError, UnicodeDecodeError): # incl. Permission/IsADirectory (V-12) return {} if not isinstance(doc, dict): return {} v = doc.get(key) return v if isinstance(v, dict) else {} def _dict_list(rel: str, key: str) -> list: """Load a top-level list `key` from artifact `rel`, tolerating a malformed file.""" try: doc = _load(rel) except (OSError, ValueError, UnicodeDecodeError): # incl. Permission/IsADirectory (V-12) return [] if not isinstance(doc, dict): return [] v = doc.get(key) return v if isinstance(v, list) else [] def _code_lists() -> dict: """code-lists.v0.json 'lists' map, or {} when absent/malformed. NOT cached across calls (re-review-3 H2): a process-lifetime cache diverged from artifact_errors()'s fresh read (enforce old rules while the health check passed the new file). A long- running service should snapshot artifacts at ITS layer, not rely on validator caching.""" return _dict_key("code-lists.v0.json", "lists") def cross_check_catalogue(pack: dict, catalogue: dict) -> list[str]: idx = _catalogue_index(catalogue) lists = _code_lists() sections_by_id = {s["id"]: s for s in (catalogue.get("sections") or []) if s.get("id")} errs: list[str] = [] for sid, fields in (pack.get("sections") or {}).items(): base, inst = _split_sid(sid) # Every section base must exist in the catalogue (re-review-2 M4): section existence # was only checked on the instance path, so a bare unknown section — empty or # carrying only x_ fields — validated clean, smuggling a parallel vocabulary past # "every field exists in the catalogue". if base not in sections_by_id: errs.append(f"unknown section not in catalogue: {_safe(sid)}") continue if inst is not None: sec = sections_by_id.get(base) if not sec.get("repeatable"): errs.append(f"section '{base}' is not repeatable — instance {_safe(sid)} is not allowed") continue # ASCII-only, no leading zero, >= 2 (sec review F3 + re-review #2): match the # value directly with a regex — NO int() (str.isdigit() accepts Unicode digits; # int() on a million-digit suffix raises ValueError under Python's digit cap). if not re.fullmatch(r"[2-9]|[1-9][0-9]+", inst): errs.append(f"bad instance suffix {_safe(sid)} — further instances are " f"'{base}:2', '{base}:3', … (no leading zeros; the first " f"instance uses the bare id)") continue if base not in (pack.get("sections") or {}): errs.append(f"instance '{sid}' present without the bare first instance '{base}'") for fid, fld in fields.items(): key = f"{sid}.{fid}" if fid.startswith("x_"): # enforce the SAME x_ token grammar the strict schema requires (re-review-3 # M3): else `x_Bad` / `x_` pass the reference validator but fail strict-only # consumers — a producer↔consumer differential. if not re.fullmatch(r"x_[a-z][a-z0-9_]*", fid): errs.append(f"extension field {_safe(key)} must match x_[a-z][a-z0-9_]*") continue # EXTENSION namespace: producer-specific, shape-checked by the # schema, ignored by catalogue membership + conformance scoring row = idx.get(f"{base}.{fid}") if row is None: errs.append(f"unknown field not in catalogue: {_safe(key)} " f"(extension fields must use the x_ prefix)") continue # code-list binding: a non-null value on an enum field must be a listed code. # Guarded on isinstance(str) (sec review F2/IR2): codes are strings, so only # a string value can match; a list/dict value would make `val not in codes` # crash on an unhashable type — the datatype check below flags non-strings. cl = row.get("code_list") val = (fld or {}).get("value") if cl and isinstance(val, str) and lists: codes = {c.get("code") for c in (lists.get(cl) or {}).get("codes", [])} if codes and val not in codes: # Do NOT echo the offending value (sec review F5/IR11 — it may carry # personal data); name the field + the allowed set instead. errs.append(f"{_safe(key)}: value is not a code in code list '{cl}' " f"(allowed: {sorted(codes)})") prov = (fld or {}).get("provenance") or {} kind = prov.get("kind") allowed = _TAG_KINDS.get(row.get("tag"), set()) if kind not in allowed: errs.append(f"{_safe(key)}: provenance kind '{kind}' incompatible with catalogue tag " f"'{row.get('tag')}' (allowed: {sorted(allowed)})") # not_collected is the HONEST-ABSENCE marker — it must carry a null value # (sec review IR1). The inverse (any null value MUST be not_collected) is NOT # enforced: a pending declare/doc field legitimately holds value:null with its # natural kind and is tracked in readiness.missing. The forgery #5 targeted — a # null value counting as a VERIFIED FACT — is closed in check_readiness, which # only counts a verified fact when its value is non-null. if kind == "not_collected" and val is not None: errs.append(f"{_safe(key)}: provenance kind 'not_collected' requires a null value " f"(it marks an absent fact, not a placeholder for a supplied value)") # A VERIFIED fact must carry a SUBSTANTIVE value — verified-null/""/[] is # incoherent (you cannot verify an absent value); caught here (re-review #5 + H1). if kind == "verified" and _is_blank(val): errs.append(f"{_safe(key)}: provenance kind 'verified' requires a substantive value " f"(an absent/empty value cannot be a verified fact — use not_collected)") # expires_at (verified provenance) must be a real date if present (re-review-2 H4) exp = prov.get("expires_at") if exp is not None and _datatype_error("date", exp): errs.append(f"{_safe(key)}: provenance.expires_at is not a valid ISO date (YYYY-MM-DD)") # evidence_hash, when present, must be a well-formed content hash (v1.0 anchor # layer, P2). The schema enforces the same pattern; this keeps the reference # validator's own answer complete for a caller that skips JSON Schema. eh = prov.get("evidence_hash") if eh is not None and not (isinstance(eh, str) and re.fullmatch(r"sha256:[0-9a-f]{64}", eh)): errs.append(f"{_safe(key)}: provenance.evidence_hash must be " f"'sha256:' + 64 lowercase hex") if kind == "verified": want = row.get("source") if want and prov.get("source") != want: errs.append(f"{_safe(key)}: verified against {_safe(prov.get('source'))} but the " f"catalogue names source '{want}'") # A verified fact's date must be a real ISO date, or freshness/staleness # cannot be judged (sec review F8). Missing date is caught by stale_facts. pdate = prov.get("date") if pdate is not None and _datatype_error("date", pdate): errs.append(f"{_safe(key)}: provenance.date is not a valid ISO date (YYYY-MM-DD)") dt = row.get("datatype") if dt and val is not None: derr = _datatype_error(dt, val) if derr: errs.append(f"{_safe(key)}: {derr}") missing = (pack.get("readiness") or {}).get("missing") if not isinstance(missing, list): errs.append("readiness.missing must be a list") else: seen_missing: set = set() for m in missing: if m in seen_missing: # duplicate entries make ambiguous chase items (re-review #34) errs.append(f"readiness.missing lists {_safe(m)} more than once") continue seen_missing.add(m) if _norm_key(m) not in idx: errs.append(f"readiness.missing references unknown field: {_safe(m)}") continue sid_part = str(m).partition(".")[0] if ":" in sid_part and sid_part not in (pack.get("sections") or {}): errs.append(f"readiness.missing references instance {_safe(sid_part)} " f"not present in the pack: {_safe(m)}") continue # `missing` means "required and NOT provided" (re-review #9) — a field that is # actually provided must not appear. "Provided" here uses the SAME _is_blank # definition as check_required (re-review-3 C3): otherwise a value:"" field is # trapped — "not provided" (H1) if omitted from missing, "but it is provided" # if listed — with no valid serialization. msid, _, mfid = str(m).partition(".") mfld = ((pack.get("sections") or {}).get(msid) or {}).get(mfid) if isinstance(mfld, dict) and not _is_blank(mfld.get("value")) \ and (mfld.get("provenance") or {}).get("kind") != "not_collected": errs.append(f"readiness.missing lists {_safe(m)} but it is provided " f"(a non-null value) — missing is for gaps, not supplied fields") return errs # ── 3 + 4. readiness semantics + entity gate ──────────────────────────────── def _iter_fields(pack: dict): # Defensive over a malformed pack (re-review-3 H4): stale_facts() is a public helper a # consumer may call before/without validate(), so skip non-dict sections/fields rather # than AttributeError on `.items()` / `.get()`. sections = pack.get("sections") if isinstance(pack, dict) else None for sid, fields in (sections or {}).items(): if not isinstance(fields, dict): continue for fid, fld in fields.items(): yield f"{sid}.{fid}", (fld if isinstance(fld, dict) else {}) def check_readiness(pack: dict, catalogue: dict) -> list[str]: idx = _catalogue_index(catalogue) r = pack.get("readiness") or {} status, level = r.get("status"), r.get("conformance_level") attested = r.get("attested") is True missing = r.get("missing") if isinstance(r.get("missing"), list) else [] errs: list[str] = [] # status coherence if status == "oven-ready" and missing: errs.append("oven-ready requires readiness.missing == [] " "(list required gaps and use status gaps-remaining)") if status == "gaps-remaining" and not missing: errs.append("gaps-remaining requires a non-empty readiness.missing list") # conformance semantics verified_count = 0 for key, fld in _iter_fields(pack): fid = key.rpartition(".")[2] if fid.startswith("x_"): continue # extension fields NEVER count toward conformance (sec review IR1): # a producer-defined x_ field with kind 'verified' is not a # catalogue-defined, source-checkable fact. row = idx.get(_norm_key(key)) kind = (fld.get("provenance") or {}).get("kind") if kind == "verified" and row is not None and not _is_blank(fld.get("value")): verified_count += 1 # a verified FACT is a real catalogue field with a SUBSTANTIVE # value (re-review #5 + re-review-2 H1: null/""/[] is not a fact) if row and row.get("tag") == "verify" and level in ("L2-verified-core", "L3-attested"): if kind == "self-declared": errs.append(f"{_safe(key)}: {level} forbids self-declared values on verify-tagged fields") if level in ("L2-verified-core", "L3-attested") and verified_count == 0: errs.append(f"{level} requires at least one verified fact; pack has none") if status == "oven-ready" and level == "L1-declared": errs.append("oven-ready cannot be conformance L1-declared") if level == "L3-attested": if not attested: errs.append("L3-attested requires attested=true") if status != "oven-ready": errs.append("L3-attested requires status oven-ready") # attestation substance if attested: ts = ((pack.get("sections") or {}).get("declarations") or {}).get("truth_statement") or {} val = ts.get("value") if not (isinstance(val, str) and val.strip()): errs.append("attested=true requires declarations.truth_statement with a non-empty value") att = r.get("attestation") or {} # Require substance, not mere truthiness (re-review #14): " " / "banana" must fail. if not (isinstance(att.get("attested_by"), str) and att.get("attested_by").strip()): errs.append("attested=true requires readiness.attestation.attested_by (a non-empty name)") at = att.get("attested_at") if not (isinstance(at, str) and _iso_datetime_error(at) is None): errs.append("attested=true requires readiness.attestation.attested_at " "(an ISO date or date-time)") # entity gate (the perimeter control — corporate forms only for readiness claims) form = str(((pack.get("entity") or {}).get("legal_form")) or "").lower() if (status == "oven-ready" or attested or level in ("L2-verified-core", "L3-attested")) \ and form not in _CORPORATE_FORMS: errs.append(f"entity gate: legal_form '{form}' cannot claim oven-ready/attested/L2+ — " "v0 covers corporate forms (ltd/llp/plc) only; sole-trader/partnership " "lending may be a regulated credit agreement (RAO art 60B; art 60C exemptions), out of v0 scope") # entity vs business_identity consistency (bi_form is unconstrained by the shape schema # — do not echo it raw, re-review-2 M1) bi_form = (((pack.get("sections") or {}).get("business_identity") or {}) .get("legal_form") or {}).get("value") if bi_form and form and str(bi_form).lower() != form: errs.append(f"entity.legal_form '{form}' != business_identity.legal_form {_safe(bi_form)}") # contradictory attestation signals (re-review-2 M6): an attestation object present while # attested is not true lets a consumer keying off the object treat the pack as attested. if r.get("attestation") and not attested: errs.append("readiness.attestation is present but readiness.attested is not true " "— contradictory signals; set attested=true or remove the attestation") return errs # ── 5. baseline requirement profile ───────────────────────────────────────── def _requirement_profiles() -> dict: """requirement-profiles.v0.json 'profiles' map, or {} when absent/malformed. Uncached (H2).""" return _dict_key("requirement-profiles.v0.json", "profiles") def _overlays() -> dict: """requirement-profiles.v0.json 'overlays' map, or {} when absent/malformed. Uncached (H2).""" return _dict_key("requirement-profiles.v0.json", "overlays") def required_fields_for(pack: dict, catalogue: dict) -> tuple[list[str], str]: """(required_fields, profile_label) for this pack: the per-product profile selected by the_ask.facility_type (else the catalogue baseline), PLUS any active overlay (e.g. sharia-v0 when the_ask.finance_basis is sharia_compliant — base overlay requirements + the chosen structure's evidence set).""" # Defensive over a malformed pack/catalogue: this is a PUBLIC helper that callers # (e.g. evidence.core_fields) invoke directly, without validate()'s structure-first # gate in front of it, so it must be total over hostile input. if not isinstance(pack, dict) or not isinstance(catalogue, dict): return [], "baseline-v0" sections = pack.get("sections") sections = sections if isinstance(sections, dict) else {} the_ask = sections.get("the_ask") ftype = ((the_ask if isinstance(the_ask, dict) else {}).get("facility_type") or {}) ftype = ftype.get("value") if isinstance(ftype, dict) else None required, label = list(catalogue.get("required_fields") or []), "baseline-v0" if ftype: for name, prof in _requirement_profiles().items(): if ftype in (prof.get("applies_to") or []): required, label = list(prof.get("required_fields") or []), name break for oname, ov in _overlays().items(): if not isinstance(ov, dict): continue sel = ov.get("selected_by") or {} sid, _, fid = str(sel.get("field", "")).partition(".") sec = sections.get(sid) fld = (sec if isinstance(sec, dict) else {}).get(fid) val = (fld if isinstance(fld, dict) else {}).get("value") sel_val = sel.get("value") if isinstance(sel, dict) else None # Require a non-null selector value AND a matching non-null pack value # (sec review F1/IR7): a malformed overlay with no `value` made None==None # true, silently activating the overlay on every pack. Match TYPE too # (re-review-2 L2): Python's 1==True/0==False would else cross types. if sel_val is not None and type(val) is type(sel_val) and val == sel_val: required = required + list(ov.get("required_fields") or []) # ponytail: structure key is sharia-specific until a second overlay exists sc = sections.get("sharia_compliance") ss = (sc if isinstance(sc, dict) else {}).get("sharia_structure") struct = (ss if isinstance(ss, dict) else {}).get("value") # struct is used as a dict key — a non-string (array/object) value passes the # broad shape schema, so guard it or `.get(struct)` raises "unhashable type" # and crashes validate() (re-review #1). A bad type is caught by the datatype check. if isinstance(struct, str): required += list((ov.get("structure_required_fields") or {}).get(struct) or []) label = f"{label}+{oname}" return required, label def check_required(pack: dict, catalogue: dict) -> list[str]: """Every required field (per the selected requirement profile) must be PROVIDED (present, value not None) or listed in readiness.missing. A required gap cannot be silently omitted.""" required, profile = required_fields_for(pack, catalogue) r = pack.get("readiness") or {} missing_set = set(r.get("missing") if isinstance(r.get("missing"), list) else []) sections = pack.get("sections") or {} errs: list[str] = [] sections_by_id = {s["id"]: s for s in (catalogue.get("sections") or []) if s.get("id")} for rid in required: sid, _, fid = rid.partition(".") # A required field applies to EVERY PRESENT INSTANCE of a repeatable section, not # just the bare first one (v0.6: sec review M7/#15). Otherwise a second guarantor # can carry one harmless field and the pack still reads oven-ready — the ':2' # instance grammar exists precisely so this is expressible. targets = [rid] if (sections_by_id.get(sid) or {}).get("repeatable"): targets += [f"{osid}.{fid}" for osid in sections if _split_sid(osid) == (sid, _split_sid(osid)[1]) and _split_sid(osid)[1] is not None] for target in targets: tsid, _, tfid = target.partition(".") fld = (sections.get(tsid) or {}).get(tfid) prov_kind = (fld.get("provenance") or {}).get("kind") if isinstance(fld, dict) else None # "Provided" means a SUBSTANTIVE value (re-review-2 H1): "", [], " " and a # not_collected marker (sec review IR1) all count as NOT provided — otherwise a # required gap can be forged with an empty token and the pack still reads oven-ready. provided = (isinstance(fld, dict) and not _is_blank(fld.get("value")) and prov_kind != "not_collected") if not provided and target not in missing_set: errs.append(f"required field '{target}' (profile {profile}) is not provided " f"(value is absent or empty) and not listed in readiness.missing") # NOT enforced: `missing` ⊆ required. Two reviews proposed it (sec review #14 / M8) on the # grounds that an optional entry "manufactures" a gaps-remaining state. Implementing it # immediately failed our own example-pack.multi fixture, which honestly declares # 'guarantor:2.guarantor_id' as outstanding — useful (it drives the chase) and not # required under the baseline profile. Listing a known optional gap UNDERSTATES readiness; # the direction that matters is hiding a required one, and `oven-ready ⇒ missing == []` # plus the required-field rule above already close that. So the SPEC text was the thing # that was wrong, and it now says required MUST appear while optional MAY (§4). return errs # ── artifact self-check (the standard's own files, cross-validated) ────────── def check_artifacts(catalogue: dict) -> list[str]: """Validate the standard's own artifacts against each other — field shape and vocabulary, code-list and reference-data bindings, requirement-profile coverage. The repeatable CI form of the adversarial audit's consistency battery.""" errs: list[str] = [] if not isinstance(catalogue, dict) or not isinstance(catalogue.get("sections"), list): return ["catalogue must be an object with a 'sections' list"] lists = _code_lists() regs = {r.get("id") for r in (_dict_list("reference-data.v0.json", "registries")) if isinstance(r, dict)} seen_sids: set = set() seen_fqs: set = set() for s in catalogue.get("sections") or []: # defensive over a malformed catalogue (re-review-3 C4/C6): this is the layer that # must convert a bad trust-root into a CLEAN refusal, so it must not itself crash. if not isinstance(s, dict): errs.append(f"catalogue section is not an object: {_safe(s)}") continue sid_raw = s.get("id") if not isinstance(sid_raw, str) or not sid_raw or not s.get("title"): errs.append(f"section missing string id/title: {_safe(sid_raw)}") elif not _ID_RE.match(sid_raw): errs.append(f"section id {sid_raw!r} is not a safe [a-z][a-z0-9_]* token") elif sid_raw.startswith("x_"): errs.append(f"section id {sid_raw!r} uses the reserved x_ extension prefix") # #17 if isinstance(sid_raw, str) and sid_raw: if sid_raw in seen_sids: # dup section ids silently overwrite in the index (#16) errs.append(f"duplicate section id: {sid_raw!r}") seen_sids.add(sid_raw) seen_fids: set = set() fields = s.get("fields") if fields is not None and not isinstance(fields, list): errs.append(f"section {_safe(sid_raw)} 'fields' is not a list") fields = [] for f in fields or []: if not isinstance(f, dict): errs.append(f"section {_safe(sid_raw)} has a non-object field: {_safe(f)}") continue fid_raw = f.get("id") fq = f"{sid_raw}.{fid_raw}" for k in ("id", "label", "tag", "datatype", "definition"): if not f.get(k): errs.append(f"{fq}: missing {k}") if isinstance(fid_raw, str) and fid_raw and not _ID_RE.match(fid_raw): errs.append(f"{fq}: field id is not a safe [a-z][a-z0-9_]* token") if isinstance(fid_raw, str) and fid_raw.startswith("x_"): errs.append(f"{fq}: field id uses the reserved x_ extension prefix") # #17 f = {k: f.get(k) for k in ("id", "label", "tag", "datatype", "definition", "code_list", "source")} # normalized, safe .get below if f.get("id") and f["id"] in seen_fids: # dup field id within a section (#16) errs.append(f"{fq}: duplicate field id within section") if f.get("id"): seen_fids.add(f["id"]) if fq in seen_fqs: errs.append(f"duplicate fully-qualified field id: {fq}") seen_fqs.add(fq) if f.get("tag") not in _TAG_KINDS: errs.append(f"{fq}: unknown tag {f.get('tag')!r}") if f.get("datatype") and f["datatype"] not in _DATATYPE_VOCAB: errs.append(f"{fq}: unknown datatype {f['datatype']!r}") # an enum field MUST name a non-empty code_list, else it is an unconstrained # string in both the reference validator and the generated strict schema (#20) if f.get("datatype") == "enum" and not f.get("code_list"): errs.append(f"{fq}: datatype 'enum' must name a code_list") # NOTE: no `and lists`/`and regs` guard (re-review-2 H2) — a referenced code # list or registry that is ABSENT (empty/missing artifact) must FAIL LOUD, not # be skipped. artifact_errors() ensures the files exist; this ensures the # catalogue's references actually resolve inside them. if f.get("code_list") and f["code_list"] not in lists: errs.append(f"{fq}: code_list '{f['code_list']}' not in code-lists") elif f.get("code_list") and not (lists.get(f["code_list"]) or {}).get("codes"): errs.append(f"{fq}: code_list '{f['code_list']}' has no codes") # #19 empty list if f.get("source") and f["source"] not in regs: errs.append(f"{fq}: source '{f['source']}' has no reference-data registry") # every code-list entry must carry a string "code" (re-review-2 L8): the strict-schema # generator does c["code"], so a malformed entry would KeyError past the generator's guard. for lname, ldef in (lists or {}).items(): for c in (ldef or {}).get("codes") or []: if not isinstance(c, dict) or not isinstance(c.get("code"), str): errs.append(f"code list '{lname}' has an entry without a string 'code'") break idx = _catalogue_index(catalogue) for rid in catalogue.get("required_fields") or []: if rid not in idx: errs.append(f"required_fields entry not in catalogue: {_safe(rid)}") profiles = _requirement_profiles() mapped: dict[str, list[str]] = {} for name, p in profiles.items(): if not isinstance(p, dict): errs.append(f"profile {_safe(name)} is not an object") continue for rid in p.get("required_fields") or []: if rid not in idx: errs.append(f"profile {name}: required field not in catalogue: {_safe(rid)}") for c in p.get("applies_to") or []: if isinstance(c, str): mapped.setdefault(c, []).append(name) for c, names in mapped.items(): if len(names) > 1: errs.append(f"facility_type '{c}' mapped by multiple profiles: {names}") # only string codes enter the set (re-review-3 C6): a None/non-str would crash sorted() ft = {c.get("code") for c in (lists.get("facility_type") or {}).get("codes", []) if isinstance(c, dict) and isinstance(c.get("code"), str)} if ft: tax = {t for t in (_dict_key("requirement-profiles.v0.json", "taxonomy_only") .get("codes") or []) if isinstance(t, str)} for c in sorted(ft - set(mapped) - tax): errs.append(f"facility_type '{c}' has no profile and is not taxonomy_only") errs += _overlay_errors(idx, lists) # re-review-3 H1: validate overlays too return errs def _overlay_errors(idx: dict, lists: dict) -> list[str]: """Validate the requirement-profile OVERLAYS (re-review-3 H1): the consistency battery never looked at them, so a selector typo silently disabled the sharia gate for every pack, and typo'd overlay required-field ids were unsatisfiable forever.""" errs: list[str] = [] sharia_structures = {c.get("code") for c in (lists.get("sharia_structure") or {}).get("codes", []) if isinstance(c, dict)} for oname, ov in _overlays().items(): if not isinstance(ov, dict): errs.append(f"overlay {_safe(oname)} is not an object") continue sel = ov.get("selected_by") or {} selfield = sel.get("field") if not (isinstance(selfield, str) and _norm_key(selfield) in idx): errs.append(f"overlay {oname}: selected_by.field {_safe(selfield)} is not a catalogue field") for rid in (ov.get("required_fields") or []): if rid not in idx: errs.append(f"overlay {oname}: required field {_safe(rid)} not in catalogue") srf = ov.get("structure_required_fields") or {} for struct, rids in (srf.items() if isinstance(srf, dict) else []): if sharia_structures and struct not in sharia_structures: errs.append(f"overlay {oname}: structure key {_safe(struct)} is not a sharia_structure code") for rid in (rids or []): if rid not in idx: errs.append(f"overlay {oname}: structure field {_safe(rid)} not in catalogue") return errs # ── disposition messages (the funder's response — spec/disposition.md) ────── def validate_disposition(msg: dict, catalogue: dict | None = None) -> list[str]: """Validate a funder disposition message. `catalogue` defaults to the bundled one. Fail-closed: never raises on hostile input (re-review-3).""" return _fail_closed(_validate_disposition_impl, msg, catalogue if catalogue is not None else _bundled_catalogue()) def _validate_disposition_impl(msg: dict, catalogue: dict) -> list[str]: """structure (disposition.schema.json), status/decline codes against the code lists, and info_requested.field_ids against the catalogue (instance-qualified ids permitted).""" try: import jsonschema # type: ignore except ImportError: return ["DEGRADED: the 'jsonschema' library is not installed — disposition " "was NOT validated. Install jsonschema; a message cannot pass without it."] if not isinstance(msg, dict): return ["disposition must be a JSON object"] # Dispositions need only the code lists (decline_reason), NOT the requirement profiles — # coupling them was an availability footgun (re-review-2 L9). arte = artifact_errors(("code-lists.v0.json",)) if arte: return arte # SCOPED catalogue check (re-review-3 H3): shape + version only, so a missing # requirement-profiles/reference-data file cannot take dispositions down (re-review-2 L9). cat_err = _catalogue_shape_errors(catalogue) if cat_err: return cat_err if _too_deep(msg) or _too_broad(msg) or _oversized_string(msg): return ["disposition exceeds the bounded size/depth/string limit — refused (re-review #6/#9)"] if _has_nonfinite(msg): return ["disposition contains a non-finite number (NaN/Infinity) — refused (re-review #8)"] # Bind the message version (re-review #12): a 999.0.0 disposition must not be validated # under the 0.2 schema. dv = msg.get("disposition_version") if isinstance(dv, str) and tuple(dv.split(".")[:2]) != _SUPPORTED_DISPOSITION_MAJMIN: return [f"disposition_version '{dv}' is not supported (validator implements " f"{'.'.join(_SUPPORTED_DISPOSITION_MAJMIN)}.x)"] schema = _load("disposition.schema.json") validator = jsonschema.Draft202012Validator(schema) structure = [f"{'/'.join(map(str, e.path))}: {e.message}" for e in validator.iter_errors(msg)] if structure: return structure # structure is a hard prerequisite (sec review IR2) errs: list[str] = [] # timestamp must be a real ISO date/date-time (re-review-2 M2): it is the ONLY # replay/freshness hook the spec offers a consumer, so "banana" must not pass. if _iso_datetime_error(msg.get("timestamp")): errs.append("timestamp is not a valid ISO date or date-time") lists = _code_lists() # status against the disposition_status code list, not only the schema enum (re-review-3 # #26): otherwise the schema and the code-list artifact can silently drift apart. scodes = {c.get("code") for c in (lists.get("disposition_status") or {}).get("codes", [])} if not scodes: errs.append("DEGRADED: code list 'disposition_status' is missing/empty") elif msg.get("status") not in scodes: errs.append(f"status {_safe(msg.get('status'))} is not a code in code list 'disposition_status'") if msg.get("status") == "declined": rc = (msg.get("declined") or {}).get("reason_code") rcodes = {c.get("code") for c in (lists.get("decline_reason") or {}).get("codes", [])} if not rcodes: # fail LOUD if the decline_reason list is missing (re-review-2 H2) errs.append("DEGRADED: code list 'decline_reason' is missing/empty — cannot check the reason") elif rc not in rcodes: errs.append(f"declined.reason_code {_safe(rc)} is not a code in code list 'decline_reason' " f"(allowed: {sorted(rcodes)})") idx = _catalogue_index(catalogue) sections_by_id = {s["id"]: s for s in (catalogue.get("sections") or []) if s.get("id")} seen_fids: set = set() for f in ((msg.get("info_requested") or {}).get("field_ids") or []): if f in seen_fids: # duplicate requested ids make ambiguous chase items (re-review-2 L4) errs.append(f"info_requested.field_ids lists {_safe(f)} more than once") continue seen_fids.add(f) if _norm_key(f) not in idx: errs.append(f"info_requested.field_ids references unknown field: {_safe(f)}") continue gerr = _instance_ref_error(f, sections_by_id) # canonical suffix + repeatability (#10) if gerr: errs.append(f"info_requested.field_ids {gerr}") return errs # ── producer identity (who assembled the pack — schema v0.5.0) ─────────────── _FCA_STATUS = {"authorised", "appointed_representative", "unregulated", "not_applicable", "not_stated"} def check_producer(pack: dict) -> list[str]: """The `producer` block is optional. When present it must name the party, and any stated FCA position must be a listed status. An appointed representative must name its principal (FSMA s.39 puts responsibility on the principal), and an authorised producer or AR should carry its Firm Reference Number so a consumer can check the FCA Register entry.""" prod = pack.get("producer") if prod is None: return [] errs: list[str] = [] if not str(prod.get("name") or "").strip(): errs.append("producer.name is required when a producer block is present") status = prod.get("fca_status") if status is not None and status not in _FCA_STATUS: errs.append(f"producer.fca_status '{status}' is not a listed status " f"(allowed: {sorted(_FCA_STATUS)})") if status == "appointed_representative" and not str(prod.get("principal_name") or "").strip(): errs.append("producer.principal_name is required when fca_status is " "appointed_representative (FSMA s.39: the principal carries " "regulatory responsibility)") frn = str(prod.get("fca_frn") or "").strip() if status in ("authorised", "appointed_representative") and not frn: # MUST, not SHOULD (sec review IR13): the validator rejects, so the spec and # message align on MUST — an unresolvable FRN defeats consumer verifiability. errs.append(f"producer.fca_frn must be provided when fca_status is '{status}' " "so a consumer can check the FCA Register") # Format-check whenever an FRN is PRESENT, not only when a status demands one # (fresh-eyes review V-10): a malformed FRN alongside an absent/unregulated status # went unchecked, and an unresolvable FRN is the whole reason the field exists. if frn and not re.fullmatch(r"[0-9]{6,7}", frn): errs.append("producer.fca_frn must be a 6–7 digit FCA Firm Reference Number") return errs # ── sharia overlay rules (opt-in — spec/sharia-structuring.md) ─────────────── def check_sharia(pack: dict, catalogue: dict) -> list[str]: """Active ONLY when the_ask.finance_basis is 'sharia_compliant' (conventional packs are untouched). Enforces: the deterministic SECTOR GATE (verified SIC codes vs the sharia_excluded_sic prefixes), the murabaha/tawarruq SEQUENCE rule (the funder must own before it sells), and S-ladder coherence (S2 = structure evidence provided; S3 = S2 + certificate + board + review date). The standard never rules on religious law — it evidences structure and records certification; the funder's own board's gates dispose. NOTE (re-review-3 H6): the sector gate fires on the SIC codes PRESENT in the pack regardless of provenance — declaring an excluded activity excludes you. It does not itself prove the SIC is verified/current; that is the consumer's diligence (§4a). It is 'deterministic' over the codes present, not a guarantee they are the applicant's true codes.""" sections = pack.get("sections") or {} basis = ((sections.get("the_ask") or {}).get("finance_basis") or {}).get("value") sc = sections.get("sharia_compliance") or {} errs: list[str] = [] if basis != "sharia_compliant": if sc: errs.append("sharia_compliance section present but the_ask.finance_basis is not " "'sharia_compliant' — set the switch or remove the section") return errs # sector gate — deterministic, over VERIFIED registry codes lists = _code_lists() prefixes = [c.get("code") for c in (lists.get("sharia_excluded_sic") or {}).get("codes", [])] sic_val = ((sections.get("business_identity") or {}).get("sic_codes") or {}).get("value") if sic_val is not None and not isinstance(sic_val, (list, str)): errs.append("sharia sector gate: business_identity.sic_codes is not a code list — " "the sector gate cannot be evaluated (refused rather than passed)") sic_list = sic_val if isinstance(sic_val, list) else ( [sic_val] if isinstance(sic_val, str) else []) for code in sic_list: norm = _norm_sic(code) if norm is None: # FAIL CLOSED (fresh-eyes review V-01): an uninterpretable code must not # silently pass a compliance gate. A zero-width space or a nested list made # the prefix match miss, and the pack validated clean. errs.append(f"sharia sector gate: SIC code {_safe(code)} is not a plain numeric " f"code — the sector gate cannot be evaluated (refused, not passed)") continue hit = next((p for p in prefixes if p and norm.startswith(p)), None) if hit: errs.append(f"sharia sector gate: SIC code {_safe(code)} matches an excluded prefix " f"— this pack cannot carry finance_basis 'sharia_compliant'") # sequence rule — ownership before onward sale. PARSE the dates (re-review-2 M9): # lexical string compare is only correct for zero-padded ISO dates; parse so a # non-padded '2026-6-1' can't mis-evaluate a structuring-compliance control. Values # are not echoed (M1). from datetime import date as _date t = (sc.get("ownership_transfer_date") or {}).get("value") s = (sc.get("sale_contract_date") or {}).get("value") if isinstance(t, str) and isinstance(s, str): try: if _date.fromisoformat(s) < _date.fromisoformat(t): errs.append("sharia sequence: sale_contract_date precedes " "ownership_transfer_date — the funder must own before it sells") except ValueError: # FAIL CLOSED (fresh-eyes review V-06): a date the gate cannot parse (e.g. a # date-time) previously skipped the ordering check silently. The datatype check # in cross_check_catalogue is the primary net, but a compliance gate must not # depend on another rule staying in place. errs.append("sharia sequence: ownership_transfer_date / sale_contract_date are " "not plain ISO dates — the ordering gate cannot be evaluated") # S-ladder coherence level = (sc.get("claimed_level") or {}).get("value") struct = (sc.get("sharia_structure") or {}).get("value") if level in ("S2-structure-evidenced", "S3-certified"): if not isinstance(struct, str) or not struct: errs.append(f"{level} requires sharia_compliance.sharia_structure to be provided") # guard struct as a dict key (re-review #1): a non-string value must not crash lookup needed = ((_overlays().get("sharia-v0") or {}) .get("structure_required_fields") or {}).get(struct if isinstance(struct, str) else None) or [] for rid in needed: rsid, _, rfid = rid.partition(".") fld = (sections.get(rsid) or {}).get(rfid) # substantive, not merely non-null (re-review-3 #16): "" / [] / " " is not evidence if not (isinstance(fld, dict) and not _is_blank(fld.get("value"))): errs.append(f"{level} requires structure evidence '{rid}' to be provided") if level == "S3-certified": for fid in ("sharia_certificate", "certifying_board", "certificate_expiry"): fld = sc.get(fid) if not (isinstance(fld, dict) and not _is_blank(fld.get("value"))): errs.append(f"S3-certified requires sharia_compliance.{fid} to be provided") # the certificate must not already be expired (re-review-3 #16): S3 asserts a live cert exp = (sc.get("certificate_expiry") or {}).get("value") if isinstance(exp, str) and _datatype_error("date", exp) is None: from datetime import date as _d asof = ((sc.get("certificate_as_of") or {}).get("value")) # compare against a supplied as_of if present; otherwise this is a structural # coherence check only (freshness vs "today" stays caller-driven, like stale_facts) if isinstance(asof, str) and _datatype_error("date", asof) is None and _d.fromisoformat(exp) < _d.fromisoformat(asof): errs.append("S3-certified: certificate_expiry is before certificate_as_of (expired)") return errs # ── pack identity (pack_ref) ───────────────────────────────────────────────── def canonical_json(obj) -> bytes: """RECOMMENDED canonical serialization for hashing: UTF-8, keys sorted, compact separators, no insignificant whitespace. Producers/consumers who compute pack identity independently of transmission SHOULD use this form so their hashes agree.""" # allow_nan=False (sec review IR4/F2): NaN/Infinity are not valid JSON — refuse to # produce a "canonical" form for a document a conforming parser would reject. # # ⚠ NOT RFC 8785. Python renders an integer-valued float as `1000.0` where # ECMAScript/Go/JCS render `1000`, so a Python and a JS/Go producer hashing the SAME # parsed structure get DIFFERENT pack_refs (fresh-eyes review V-03 — confirmed). This # form is safe only where every party shares this exact implementation, or where all # parties hash the exact transmitted bytes (the normative rule, spec/disposition.md). # For genuine cross-language agreement use `evidence.jcs` / `evidence.pack_ref_jcs` # (RFC 8785); adopting it as the pack identity is the v1.0 boundary (design §4). return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False, allow_nan=False).encode("utf-8") def pack_ref(pack) -> str: """'sha256:' pack identity. NORMATIVE rule: identity is the SHA-256 of the exact UTF-8 bytes of the pack document AS TRANSMITTED — a re-serialized pack is a different identity. Pass bytes to hash a transmitted document; pass a dict to hash its canonical_json form.""" import hashlib data = pack if isinstance(pack, (bytes, bytearray)) else canonical_json(pack) return "sha256:" + hashlib.sha256(bytes(data)).hexdigest() # ── borrowing-base certificates (continuous mode — spec/borrowing-base.md) ─── def validate_bbc(msg: dict) -> list[str]: """Validate a borrowing-base certificate. Fail-closed: never raises (re-review-3).""" return _fail_closed(_validate_bbc_impl, msg) def _validate_bbc_impl(msg: dict) -> list[str]: """structure + the arithmetic rules (eligible <= gross; availability <= eligible x advance_rate; drawn <= availability; headroom reconciliation; non-negativity).""" try: import jsonschema # type: ignore except ImportError: return ["DEGRADED: the 'jsonschema' library is not installed — certificate " "was NOT validated."] if not isinstance(msg, dict): return ["certificate must be a JSON object"] if _too_deep(msg) or _too_broad(msg) or _oversized_string(msg): # same bounds (re-review-2 H3/#9) return ["certificate exceeds the bounded size/depth/string limit — refused"] if _has_nonfinite(msg): # A NaN slips past jsonschema "number" and evades every arithmetic comparison # (NaN comparisons are all false) — reject it up front (re-review #8). return ["certificate contains a non-finite number (NaN/Infinity) — refused (re-review #8)"] cvv = msg.get("certificate_version") if isinstance(cvv, str) and tuple(cvv.split(".")[:2]) != _SUPPORTED_CERTIFICATE_MAJMIN: return [f"certificate_version '{cvv}' is not supported (validator implements " f"{'.'.join(_SUPPORTED_CERTIFICATE_MAJMIN)}.x)"] # re-review #12 schema = _load("borrowing-base-certificate.schema.json") validator = jsonschema.Draft202012Validator(schema) structure = [f"{'/'.join(map(str, e.path))}: {e.message}" for e in validator.iter_errors(msg)] if structure: return structure # structure is a hard prerequisite (sec review IR2) errs: list[str] = [] gross = msg.get("gross_receivables") eligible = msg.get("eligible_receivables") rate = msg.get("advance_rate") avail = msg.get("availability") drawn = msg.get("drawn_balance") headroom = msg.get("headroom") num = _is_finite_number # rejects NaN/Infinity so arithmetic comparisons are meaningful (F2) # MONEY IS DECIMAL, NOT BINARY FLOAT (sec review #27/#34). Every monetary comparison below # runs in Decimal, quantized to the penny, so: # * 0.1 + 0.2 == 0.3 holds, as it must in a lending standard; # * large values keep cent precision (a £100m facility loses pennies in float64); # * the arbitrary `+ 0.01` tolerance is GONE. It silently accepted exactly one penny of # overstatement (the comparison was `>`, not `>=`) — a fudge with no policy behind it. # ROUNDING POLICY: both sides are rounded half-up to 2dp and compared exactly, so a # certificate that reconciles to the penny passes and one that does not, fails. gross, eligible, avail, drawn, headroom = (_money(x) for x in (gross, eligible, avail, drawn, headroom)) rate = _money(rate, places=6) # a rate is not money; keep more places # Do NOT echo the monetary values into errors (sec review F5/IR11) — name the rule. # Non-negativity (re-review-2 H3): a negative amount/rate is nonsense, and a negative # rate makes the availability rule pass a negative availability. for name in ("gross_receivables", "eligible_receivables", "advance_rate", "availability", "drawn_balance", "headroom"): v = msg.get(name) if num(v) and v < 0: errs.append(f"{name} must not be negative") if rate is not None and rate > 1: errs.append("advance_rate must be a fraction in [0,1]") if gross is not None and eligible is not None and eligible > gross: errs.append("eligible_receivables exceeds gross_receivables") # `availability` is the TOTAL for the facility. On a receivables-only certificate that # total is the receivables line, so it must not exceed eligible x rate. On a MULTI-ASSET # base (ABL) the total legitimately exceeds the receivables line, and the # sum-of-lines rule below is the one that applies — so gate this single-line check on # there being no inventory line, or it fires falsely on every ABL certificate. _has_inventory_line = any(msg.get(k) is not None for k in ("gross_inventory", "eligible_inventory", "inventory_advance_rate", "inventory_availability")) if not _has_inventory_line and None not in (eligible, rate, avail) \ and avail > _q(eligible * rate): errs.append("availability exceeds eligible_receivables x advance_rate") # The decision-relevant invariant: you cannot draw more than is available (re-review-2 H3). if avail is not None and drawn is not None and drawn > avail: errs.append("drawn_balance exceeds availability (the facility is overdrawn)") # headroom must reconcile to availability - drawn_balance. if None not in (avail, drawn, headroom) and headroom != _q(avail - drawn): errs.append("headroom does not reconcile to availability - drawn_balance") # ineligible reconciliation (re-review-3 H5): overstating eligible is THE borrowing-base # fraud — eligible must equal gross minus the sum of the ineligible buckets. inelig = msg.get("ineligible") if isinstance(inelig, dict) and gross is not None and eligible is not None: buckets = [_money(v) for v in inelig.values() if num(v)] if len(buckets) == len(inelig) and eligible != _q(gross - sum(buckets, _D(0))): errs.append("eligible_receivables does not reconcile to gross minus the ineligible buckets") # ── inventory line (ABL: a borrowing base can carry more than one asset class) ── # Entirely OPTIONAL: a receivables-only certificate omits these and the rules above # remain the whole story. g_inv = _money(msg.get("gross_inventory")) e_inv = _money(msg.get("eligible_inventory")) inv_rate = _money(msg.get("inventory_advance_rate"), places=6) inv_avail = _money(msg.get("inventory_availability")) inv_inelig = msg.get("ineligible_inventory") if g_inv is not None and e_inv is not None and e_inv > g_inv: errs.append("eligible_inventory exceeds gross_inventory") if isinstance(inv_inelig, dict) and g_inv is not None and e_inv is not None: b = [_money(v) for v in inv_inelig.values() if num(v)] if len(b) == len(inv_inelig) and e_inv != _q(g_inv - sum(b, _D(0))): errs.append("eligible_inventory does not reconcile to gross minus its ineligible buckets") if None not in (e_inv, inv_rate, inv_avail) and inv_avail > _q(e_inv * inv_rate): errs.append("inventory_availability exceeds eligible_inventory x inventory_advance_rate") # Total availability is the sum of the asset lines. Stated separately by the certificate, # so the two must agree — an overstated total is the multi-asset analogue of the # single-line overstatement above. if None not in (avail, inv_avail, eligible, rate): if avail > _q(eligible * rate) + inv_avail: errs.append("availability exceeds the receivables line plus the inventory line") # An inventory line that names an advance rate but no NOLV basis is a soft gap worth # naming: inventory advance rates are ordinarily struck against appraised NOLV. if inv_rate is not None and msg.get("inventory_nolv_rate") is None: errs.append("inventory_advance_rate is stated without inventory_nolv_rate — an " "inventory advance rate is ordinarily struck against appraised NOLV") # period/certification dates (re-review-3 H5): parity with the disposition timestamp check for name in ("period_start", "period_end", "certified_at"): d = msg.get(name) if d is not None and _iso_datetime_error(d): errs.append(f"{name} is not a valid ISO date or date-time") ps, pe = msg.get("period_start"), msg.get("period_end") if isinstance(ps, str) and isinstance(pe, str) \ and _datatype_error("date", ps) is None and _datatype_error("date", pe) is None: from datetime import date as _d if _d.fromisoformat(pe) < _d.fromisoformat(ps): errs.append("period_end is before period_start") return errs # ── conformance vectors (published test cases — conformance-vectors.v0.json) ─ def run_vectors(catalogue: dict) -> list[str]: """Run every published conformance vector; returns mismatches (empty = all behave as documented). Implementers can run the same file against their own validators.""" try: doc = _load("conformance-vectors.v0.json") except FileNotFoundError: # do NOT fail-open (re-review-3 #44): "0 vectors, all passed" would be a false green return ["conformance-vectors.v0.json is missing — cannot confirm vector behaviour"] except ValueError as e: return [f"conformance-vectors.v0.json is unparseable ({e})"] _runners = {"pack": lambda d: validate(d, catalogue), "disposition": lambda d: validate_disposition(d, catalogue), "bbc": lambda d: validate_bbc(d)} errs: list[str] = [] for v in doc.get("vectors") or []: vid, kind = v.get("id"), v.get("kind", "pack") if "document" not in v: # a malformed vector is a vector-file bug, reported not crashed (L11) errs.append(f"vector {vid}: missing 'document'") continue runner = _runners.get(kind) if runner is None: errs.append(f"vector {vid}: unknown kind {kind!r}") continue got = runner(v["document"]) if v.get("expect") == "valid" and got: errs.append(f"vector {vid}: expected valid, got errors: {got[:2]}") if v.get("expect") == "invalid": if not got: errs.append(f"vector {vid}: expected invalid, validated clean") elif v.get("error_contains") and not any(v["error_contains"] in e for e in got): errs.append(f"vector {vid}: errors lack '{v['error_contains']}': {got[:2]}") return errs # ── freshness helper (NOT part of validate — keeps validation clock-free) ───── def stale_facts(pack: dict, as_of: str, max_age_days: int = 90) -> list[str]: """'sec.field' ids whose VERIFIED provenance `date` is not fresh as of `as_of` (ISO date). The caller supplies 'today' so validate() stays pure. A fact is NOT fresh when its date is missing/unparseable (sec review F8), older than max_age_days, OR dated in the FUTURE (re-review #33 — a producer must not date a verification forward to make it fresh forever). Raises ValueError on invalid `as_of` or negative `max_age_days` (re-review #31/#32): a policy typo must not silently report 'nothing stale' (fail-open).""" from datetime import date try: ref = date.fromisoformat(str(as_of)) except (TypeError, ValueError): raise ValueError(f"stale_facts: as_of {as_of!r} is not an ISO date (YYYY-MM-DD)") if not isinstance(max_age_days, int) or isinstance(max_age_days, bool) or max_age_days < 0: raise ValueError(f"stale_facts: max_age_days must be a non-negative int, got {max_age_days!r}") stale: list[str] = [] for key, fld in _iter_fields(pack): if key.rpartition(".")[2].startswith("x_"): continue # x_ provenance is meaningless (re-review-2 L7); don't report it as stale prov = fld.get("provenance") or {} if prov.get("kind") != "verified": continue # An expired verification is not fresh, regardless of how recently it was taken (H4). exp = prov.get("expires_at") if isinstance(exp, str) and _datatype_error("date", exp) is None: if date.fromisoformat(exp) < ref: stale.append(key) continue try: when = date.fromisoformat(str(prov.get("date"))) except (TypeError, ValueError): stale.append(key) # a verified fact with no/invalid date cannot be shown fresh continue age = (ref - when).days if age > max_age_days or age < 0: # too old OR dated in the future stale.append(key) return stale def _version_errors(pack: dict, catalogue: dict) -> list[str]: """Reject a pack whose declared versions differ from what this validator+catalogue implement (sec review IR3 + re-review #4). During 0.x, bind on major.minor: validating a schema_version 999.0.0 (or a different catalogue major.minor) under these artifacts would apply the wrong rules and mask a version-confusion / downgrade attack. Also bind the CALLER-SUPPLIED catalogue: a mismatched/stale catalogue argument (weaker tags or required fields) must not silently validate a pack that claims the supported version.""" errs: list[str] = [] sv = pack.get("schema_version") if isinstance(sv, str) and tuple(sv.split(".")[:2]) != _SUPPORTED_SCHEMA_MAJMIN: # _safe the version echo (re-review-3 #42): it is attacker-controlled and reflected # before structure validation runs. errs.append(f"schema_version {_safe(sv)} is not supported by this validator " f"(supports {'.'.join(_SUPPORTED_SCHEMA_MAJMIN)}.x) — refuse rather " f"than validate under a different schema version") cv = pack.get("catalogue_version") if isinstance(cv, str) and tuple(cv.split(".")[:2]) != _SUPPORTED_CATALOGUE_MAJMIN: errs.append(f"catalogue_version {_safe(cv)} is not supported by this validator " f"(supports {'.'.join(_SUPPORTED_CATALOGUE_MAJMIN)}.x)") cat_v = (catalogue or {}).get("catalogue_version") if not isinstance(cat_v, str) or tuple(cat_v.split(".")[:2]) != _SUPPORTED_CATALOGUE_MAJMIN: errs.append(f"the supplied catalogue is version {cat_v!r}, not the " f"{'.'.join(_SUPPORTED_CATALOGUE_MAJMIN)}.x this validator implements — " f"refuse rather than validate against an unexpected catalogue") elif isinstance(cv, str) and cv.split(".")[:2] != cat_v.split(".")[:2]: # major.minor, NOT full-string (re-review-2 M5): a catalogue PATCH (0.10.0 -> 0.10.1) # must not invalidate every extant 0.10.x pack. errs.append(f"pack catalogue_version '{cv}' does not match the supplied catalogue " f"'{cat_v}' at major.minor — the pack was assembled against a different catalogue") return errs def catalogue_digest(catalogue: dict) -> str: """`sha256:` + hex over the canonical bytes of a catalogue — a pin for the trust root. The catalogue decides which fields exist, which are required and which are verify-tagged, so a swapped or weakened one can bless a gutted pack while still declaring a supported version. Publish this digest alongside a validator deployment and pass it to `validate` as `expect_catalogue_digest` to refuse anything else. Stable across re-serialization (keys sorted, no insignificant whitespace). It shares `canonical_json`'s caveat — not RFC 8785 — so both sides must compute it with this implementation; `evidence.jcs` is the cross-language form when that matters. """ import hashlib return "sha256:" + hashlib.sha256(canonical_json(catalogue)).hexdigest() def _catalogue_shape_errors(catalogue) -> list[str]: """Minimal catalogue shape + version check — enough to index field ids safely, WITHOUT requiring the profile/registry artifacts (re-review-3 H3): used by validate_disposition, which needs only field-id resolution and must not be coupled to requirement-profiles / reference-data availability (re-review-2 L9).""" if not isinstance(catalogue, dict): return ["catalogue must be a JSON object"] if not isinstance(catalogue.get("sections"), list): return ["catalogue.sections must be a list"] cat_v = catalogue.get("catalogue_version") if not isinstance(cat_v, str) or tuple(cat_v.split(".")[:2]) != _SUPPORTED_CATALOGUE_MAJMIN: return [f"the supplied catalogue is version {cat_v!r}, not the " f"{'.'.join(_SUPPORTED_CATALOGUE_MAJMIN)}.x this validator implements"] return [] def _catalogue_errors(catalogue: dict) -> list[str]: """Refuse a malformed/incoherent CALLER-SUPPLIED catalogue (re-review-2 H2/M3): the catalogue is a trust root, so run the FULL artifact self-check on it and refuse rather than KeyError deep inside a semantic gate or bless a pack against a gutted catalogue. Used by validate() (packs need the profile/overlay coverage check_artifacts provides).""" shape = _catalogue_shape_errors(catalogue) if shape: return shape return [f"DEGRADED (catalogue): {e}" for e in check_artifacts(catalogue)] @lru_cache(maxsize=1) def _bundled_catalogue() -> dict: return _load("field-catalogue.v0.json") def _fail_closed(fn, *args) -> list[str]: """Run a validator body so it can NEVER raise on hostile input (re-review-3 #6/#12/ #13/#25/#29/#57): any unexpected exception (malformed artifact/schema, cyclic or non-string-keyed dict, library error) becomes a DEGRADED error — the pack is refused, not blessed, and the caller never sees a traceback. Also caps the error list so a pathologically-wide invalid document can't return an unbounded response (#11).""" try: errs = fn(*args) except RecursionError: return ["DEGRADED: input too deeply nested / cyclic (RecursionError) — refused"] except Exception as e: # noqa: BLE001 — fail CLOSED, deliberately broad return [f"DEGRADED: internal validator error ({type(e).__name__}) — refusing to bless"] if len(errs) > _MAX_ERRORS: return errs[:_MAX_ERRORS] + [f"… ({len(errs) - _MAX_ERRORS} more errors suppressed; " f"error list capped at {_MAX_ERRORS})"] return errs def validate(pack: dict, catalogue: dict | None = None, *, expect_catalogue_digest: str | None = None) -> list[str]: """Validate a pack. `catalogue` defaults to the BUNDLED field catalogue — the trusted in-process constant (re-review-3 #1). Pass a catalogue only if you vouch for it as a trust root; a malformed/incoherent one is refused, but a well-formed one you supply IS taken as authoritative, so do not pass attacker-influenced catalogues.""" return _fail_closed(_validate_impl, pack, catalogue, expect_catalogue_digest) def _validate_impl(pack: dict, catalogue: dict | None, expect_catalogue_digest: str | None = None) -> list[str]: if not isinstance(pack, dict): return ["pack must be a JSON object"] if catalogue is None: catalogue = _bundled_catalogue() # Fail LOUD if the validator's own trust-root artifacts are broken (sec review F1/IR7). arte = artifact_errors() if arte: return arte if _too_deep(pack): return [f"pack nesting exceeds the bounded limit ({_MAX_DEPTH}) — refused unparsed " "(sec review F6: bounded-input guard)"] if _too_broad(pack): return [f"pack has more than {_MAX_NODES} members — refused (re-review #6: breadth guard)"] if _oversized_string(pack): return [f"pack contains a string longer than {_MAX_STR} chars — refused (re-review-3 #9)"] if _has_lone_surrogate(pack): return ["pack contains a lone UTF-16 surrogate — refused (re-review-3 C2): it is not " "interoperable JSON and cannot be hashed for pack_ref"] _nn = _first_non_nfc(pack) if _nn is not None: return [f"pack contains text that is not in Unicode NFC form ({_safe(_nn)}) — its " f"canonical bytes, and therefore its pack_ref and evidence_hash, will differ " f"from a producer that emits NFC for the same content. Normalize to NFC " f"(v0.6: sec review #50)"] if _has_nonfinite(pack): return ["pack contains a non-finite number (NaN/Infinity), which is not valid JSON " "(sec review F2 + re-review #8) — refused; parse untrusted bytes via load_document()"] # The supplied catalogue is a trust root — refuse a malformed/incoherent one up front # (re-review-2 H2/M3) so no semantic gate silently degrades or KeyErrors on it. cat_err = _catalogue_errors(catalogue) if cat_err: return cat_err # OPT-IN trust-root pin (v0.6: sec review M8/#1). The validator cannot know which # catalogue the caller trusts, so it offers the means to say so rather than guessing. if expect_catalogue_digest is not None: actual = catalogue_digest(catalogue) if actual != expect_catalogue_digest: return [f"catalogue digest {actual} does not match the pinned " f"{_safe(expect_catalogue_digest)} — refusing to validate against a " f"catalogue the caller did not authorise"] version = _version_errors(pack, catalogue) if version: return version # Structural validity is a HARD prerequisite (sec review IR2): the semantic gates # below assume dict-shaped sections/fields and would crash or mis-judge malformed # input, so a structure failure short-circuits before any semantic check runs. structure = validate_structure(pack) if structure: return structure # cross_check emits ONE error per bad field, so on a wide pack it can run to thousands # and push the readiness / entity-gate / sharia refusals past the global error cap # (fresh-eyes review V-07 — a caller filtering for "entity gate" would miss it). # Cap this stage alone; the semantic gates are bounded and always survive intact. field_errs = cross_check_catalogue(pack, catalogue) if len(field_errs) > _MAX_STAGE_ERRORS: field_errs = field_errs[:_MAX_STAGE_ERRORS] + [ f"… ({len(field_errs) - _MAX_STAGE_ERRORS} further field errors suppressed)"] return (field_errs + check_readiness(pack, catalogue) + check_required(pack, catalogue) + check_sharia(pack, catalogue) + check_producer(pack)) if __name__ == "__main__": # The self-check asserts on invariants — never let `python -O` (which strips # assert) run it and report a hollow pass (sec review F10). if not __debug__: raise SystemExit("run the self-check WITHOUT -O (assert stripping makes it hollow)") catalogue = _load("field-catalogue.v0.json") example = _load("examples/example-pack.ready.json") errors = validate(example, catalogue) assert not errors, "example pack should be valid, got:\n" + "\n".join(errors) def _clone() -> dict: return json.loads(json.dumps(example)) # invented field -> caught bad = _clone() bad["sections"]["business_identity"]["not_a_real_field"] = { "value": 1, "provenance": {"kind": "self-declared"}} assert any("not_a_real_field" in e for e in validate(bad, catalogue)) # oven-ready with a non-empty missing list -> caught bad = _clone() bad["readiness"]["missing"] = ["business_identity.lei"] assert any("missing == []" in e for e in validate(bad, catalogue)) # all-self-declared pack claiming L3-attested/oven-ready -> caught (multiple rules) bad = _clone() for _, fld in _iter_fields(bad): fld["provenance"] = {"kind": "self-declared"} bad["readiness"]["conformance_level"] = "L3-attested" errs = validate(bad, catalogue) assert any("at least one verified fact" in e for e in errs), errs assert any("forbids self-declared" in e for e in errs), errs # sole_trader claiming oven-ready -> entity gate fires bad = _clone() bad["entity"]["legal_form"] = "sole_trader" assert any("entity gate" in e for e in validate(bad, catalogue)) # attested without a real truth statement / attestation object -> caught bad = _clone() bad["sections"]["declarations"]["truth_statement"]["value"] = "" assert any("truth_statement" in e for e in validate(bad, catalogue)) bad = _clone() del bad["readiness"]["attestation"] assert any("attestation" in e for e in validate(bad, catalogue)) # verified against the wrong source -> caught bad = _clone() bad["sections"]["business_identity"]["company_name"]["provenance"]["source"] = "somewhere_else" assert any("catalogue names source" in e for e in validate(bad, catalogue)) # doc-tagged field with self-declared provenance -> tag compatibility fires bad = _clone() bad["sections"]["financials"]["statutory_accounts"]["provenance"] = {"kind": "self-declared"} assert any("incompatible with catalogue tag" in e for e in validate(bad, catalogue)) # required field not provided AND not declared in missing -> caught bad = _clone() bad["sections"]["banking"]["bank_statements"]["value"] = None assert any("bank_statements" in e and "not provided" in e for e in validate(bad, catalogue)), \ validate(bad, catalogue) # x_ extension field -> permitted (shape-checked only) ok = _clone() ok["sections"]["the_ask"]["x_internal_score"] = { "value": 0.87, "provenance": {"kind": "self-declared"}} assert not validate(ok, catalogue), validate(ok, catalogue) # non-prefixed unknown field -> still rejected bad = _clone() bad["sections"]["the_ask"]["internal_score"] = { "value": 1, "provenance": {"kind": "self-declared"}} assert any("unknown field" in e for e in validate(bad, catalogue)) # code-list binding: bad enum value -> caught bad = _clone() bad["sections"]["guarantor"]["guarantee_type"] = { "value": "handshake", "provenance": {"kind": "self-declared"}} assert any("code list 'guarantee_type'" in e for e in validate(bad, catalogue)), \ validate(bad, catalogue) # requirement profiles: an invoice-finance pack (facility_type set) claiming # oven-ready WITHOUT the debtor book -> the per-product profile fires bad = _clone() bad["sections"]["the_ask"]["facility_type"] = { "value": "invoice_discounting", "provenance": {"kind": "self-declared"}} errs = validate(bad, catalogue) assert any("aged_debtor_ledger" in e and "invoice-finance-v0" in e for e in errs), errs # taxonomy-only facility type -> falls back to baseline (still valid) # (only 'other' remains taxonomy-only since profiles 0.2.0) ok = _clone() ok["sections"]["the_ask"]["facility_type"] = { "value": "other", "provenance": {"kind": "self-declared"}} assert not validate(ok, catalogue), validate(ok, catalogue) # trade finance now has a real profile: a pack without confirmed POs cannot be oven-ready bad = _clone() bad["sections"]["the_ask"]["facility_type"] = { "value": "trade_finance", "provenance": {"kind": "self-declared"}} errs = validate(bad, catalogue) assert any("purchase_orders" in e and "trade-finance-v0" in e for e in errs), errs # repeatable instances: the two-guarantor example validates (per-instance provenance) multi = _load("examples/example-pack.multi.json") assert not validate(multi, catalogue), validate(multi, catalogue) # instance of a non-repeatable section -> rejected bad = _clone() bad["sections"]["the_ask:2"] = {"amount": {"value": 1, "provenance": {"kind": "self-declared"}}} assert any("not repeatable" in e for e in validate(bad, catalogue)) # malformed instance suffix -> rejected (first instance is the bare id) bad = _clone() bad["sections"]["guarantor:one"] = { "guarantor_name": {"value": "X", "provenance": {"kind": "self-declared"}}} assert any("bad instance suffix" in e for e in validate(bad, catalogue)) # datatype enforcement: monetary 'banana' -> caught bad = _clone() bad["sections"]["the_ask"]["amount"] = {"value": "banana", "provenance": {"kind": "self-declared"}} assert any("expected a number" in e for e in validate(bad, catalogue)) # ratio convention: 2500 as a proportion -> caught (0.7, never 70) bad = _clone() bad["sections"].setdefault("people", {})["shareholding_percent"] = { "value": 2500, "provenance": {"kind": "self-declared"}} assert any("decimal fraction" in e for e in validate(bad, catalogue)) # zero-padded instance suffix (alias of :2) -> rejected bad = _clone() bad["sections"]["guarantor:02"] = { "guarantor_name": {"value": "Z", "provenance": {"kind": "self-declared"}}} assert any("bad instance suffix" in e for e in validate(bad, catalogue)) # instance without the bare first instance -> rejected bad = _clone() bad["sections"].pop("guarantor", None) bad["sections"]["guarantor:2"] = { "guarantor_name": {"value": "Z", "provenance": {"kind": "self-declared"}}} assert any("without the bare first instance" in e for e in validate(bad, catalogue)) # missing[] citing an instance not present in the pack -> rejected bad = _clone() bad["readiness"]["status"] = "gaps-remaining" bad["readiness"]["missing"] = ["guarantor:9.guarantor_id"] assert any("not present in the pack" in e for e in validate(bad, catalogue)) # the standard's own artifacts cross-validate assert not check_artifacts(catalogue), check_artifacts(catalogue) # generated strict schema: REQUIRED (re-review-3 #46 — don't silently pass without the # advertised companion schema; regenerate via generate_strict_schema.py if missing) strict = _load("deal-package.strict.schema.json") import jsonschema # type: ignore _sv = jsonschema.Draft202012Validator(strict) assert not list(_sv.iter_errors(example)), \ [e.message for e in _sv.iter_errors(example)][:3] assert not list(_sv.iter_errors(multi)), \ [e.message for e in _sv.iter_errors(multi)][:3] garbage = _clone() garbage["sections"]["totally_invented_section"] = { "fake": {"value": 1, "provenance": {"kind": "self-declared"}}} assert list(_sv.iter_errors(garbage)), \ "strict schema must reject invented sections" # disposition: info_requested with instance-qualified field ids -> valid GOOD_REF = "sha256:" + "0" * 64 # matches the tightened pack_ref pattern (sec review IR4) disp = {"standard": "lending-deal-package-readiness/disposition", "disposition_version": "0.2.0", "pack_ref": GOOD_REF, "status": "info_requested", "timestamp": "2026-07-18T09:00:00Z", "info_requested": {"field_ids": ["financials.statutory_accounts", "guarantor:2.guarantor_id"]}} assert not validate_disposition(disp, catalogue), validate_disposition(disp, catalogue) # disposition: unknown requested field -> caught bad_d = dict(disp) bad_d["info_requested"] = {"field_ids": ["financials.not_a_field"]} assert any("unknown field" in e for e in validate_disposition(bad_d, catalogue)) # disposition: declined without a listed reason code -> caught bad_d = {"standard": "lending-deal-package-readiness/disposition", "disposition_version": "0.2.0", "pack_ref": GOOD_REF, "status": "declined", "timestamp": "2026-07-18T09:00:00Z", "declined": {"reason_code": "vibes"}} assert any("decline_reason" in e for e in validate_disposition(bad_d, catalogue)) # disposition: empty pack_ref -> rejected (minLength) bad_d = {"standard": "lending-deal-package-readiness/disposition", "disposition_version": "0.2.0", "pack_ref": "", "status": "received", "timestamp": "2026-07-19T09:00:00Z"} assert validate_disposition(bad_d, catalogue), "empty pack_ref must fail" # disposition: stray declined block on offer_issued -> rejected (exclusivity) bad_d = {"standard": "lending-deal-package-readiness/disposition", "disposition_version": "0.2.0", "pack_ref": GOOD_REF, "status": "offer_issued", "timestamp": "2026-07-19T09:00:00Z", "declined": {"reason_code": "outside_credit_appetite"}} assert validate_disposition(bad_d, catalogue), "stray declined block must fail" # sharia overlay: the murabaha fixture validates (S2, gaps-remaining) sharia = _load("examples/example-pack.sharia.json") assert not validate(sharia, catalogue), validate(sharia, catalogue) # sector gate: excluded SIC + the switch -> gate fires bad = json.loads(json.dumps(sharia)) bad["sections"]["business_identity"]["sic_codes"] = { "value": ["92000"], "provenance": {"kind": "verified", "source": "companies_house", "via": "api", "date": "2026-07-10"}} assert any("sector gate" in e for e in validate(bad, catalogue)) # sequence: sale before ownership -> caught bad = json.loads(json.dumps(sharia)) bad["sections"]["sharia_compliance"]["sale_contract_date"]["value"] = "2026-06-01" assert any("must own before it sells" in e for e in validate(bad, catalogue)) # S2 claim without structure evidence -> caught bad = json.loads(json.dumps(sharia)) bad["sections"]["sharia_compliance"]["disclosed_markup"]["value"] = None errs = validate(bad, catalogue) assert any("structure evidence" in e and "disclosed_markup" in e for e in errs), errs # S3 claim without the certificate -> caught bad = json.loads(json.dumps(sharia)) bad["sections"]["sharia_compliance"]["claimed_level"]["value"] = "S3-certified" assert any("S3-certified requires" in e for e in validate(bad, catalogue)) # sharia section without the switch -> caught (conventional packs untouched otherwise) bad = json.loads(json.dumps(sharia)) del bad["sections"]["the_ask"]["finance_basis"] assert any("finance_basis" in e for e in validate(bad, catalogue)) # producer identity: the ready example names its producer and validates assert example.get("producer"), "the ready example should carry a producer block" # producer without a name -> caught (structure gates it: producer.name is required # + minLength 1, so the structural error fires before check_producer's semantic guard) bad = _clone() bad["producer"] = {"role": "origination platform"} assert any("name" in e and "required" in e for e in validate(bad, catalogue)) # appointed representative without a principal -> caught bad = _clone() bad["producer"] = {"name": "Some Broker Ltd", "fca_status": "appointed_representative", "fca_frn": "123456"} assert any("principal_name is required" in e for e in validate(bad, catalogue)) # authorised producer without an FRN -> caught (MUST, sec review IR13) bad = _clone() bad["producer"] = {"name": "Some Lender Ltd", "fca_status": "authorised"} assert any("fca_frn must be provided" in e for e in validate(bad, catalogue)) # pack identity: canonical serialization is stable; bytes and dict forms agree assert pack_ref(example).startswith("sha256:") assert pack_ref(example) == pack_ref(json.loads(canonical_json(example).decode("utf-8"))) assert pack_ref(canonical_json(example)) == pack_ref(example) # borrowing-base certificate: coherent -> valid; overstated -> both rules fire bbc = {"standard": "lending-deal-package-readiness/borrowing-base-certificate", "certificate_version": "0.1.0", "facility_ref": "FAC-001", "period_start": "2026-06-01", "period_end": "2026-06-30", "gross_receivables": 500000, "ineligible": {"aged_over_limit": 40000, "contra": 10000}, "eligible_receivables": 450000, "advance_rate": 0.85, "availability": 382500, "drawn_balance": 300000, "headroom": 82500, "certified_by": "B. Keeper, finance director", "certified_at": "2026-07-02"} assert not validate_bbc(bbc), validate_bbc(bbc) bad_c = dict(bbc, eligible_receivables=600000) assert any("exceeds gross" in e for e in validate_bbc(bad_c)) bad_c = dict(bbc, availability=450000) assert any("advance_rate" in e for e in validate_bbc(bad_c)) # published conformance vectors all behave as documented assert not run_vectors(catalogue), run_vectors(catalogue) # freshness helper (separate from validate; clock supplied by caller) assert stale_facts(example, "2026-07-14", max_age_days=90) == [] assert any("company_name" in s for s in stale_facts(example, "2027-01-01", max_age_days=90)) # ── security-review hardening cases ───────────────────────────────────── # IR1(A): not_collected must carry a null value — a smuggled non-null value is # rejected (readiness-forgery: "no fact here" while passing a value to the gate) bad = _clone() bad["sections"]["business_identity"]["company_name"] = { "value": "Ghost Ltd", "provenance": {"kind": "not_collected"}} assert any("not_collected" in e and "null" in e for e in validate(bad, catalogue)), \ validate(bad, catalogue) # IR1(B): an x_ extension field with kind 'verified' does NOT count as a verified # fact — a pack whose ONLY verified provenance is on x_ cannot reach L2 bad = _clone() for _, fld in _iter_fields(bad): fld["provenance"] = {"kind": "self-declared"} bad["readiness"]["conformance_level"] = "L1-declared" bad["readiness"]["status"] = "gaps-remaining" bad["readiness"]["missing"] = ["business_identity.lei"] bad["sections"]["the_ask"]["x_forged"] = { "value": 1, "provenance": {"kind": "verified", "source": "companies_house", "via": "api", "date": "2026-07-10"}} bad["readiness"]["conformance_level"] = "L2-verified-core" assert any("at least one verified fact" in e for e in validate(bad, catalogue)), \ "x_ verified extension must not satisfy L2" # IR2: a non-object pack is refused, not crashed assert validate([], catalogue) == ["pack must be a JSON object"] assert validate("nope", catalogue) == ["pack must be a JSON object"] # IR2: an enum field carrying a list value is refused by structure, never crashes # the code-list membership check (`val not in codes` on an unhashable type) bad = _clone() bad["sections"]["guarantor"]["guarantee_type"] = { "value": ["a", "b"], "provenance": {"kind": "self-declared"}} assert validate(bad, catalogue), "list-valued enum must be rejected (not crash)" # IR3: an unsupported schema_version is refused before any rule runs bad = _clone() bad["schema_version"] = "999.0.0" assert any("not supported by this validator" in e for e in validate(bad, catalogue)) # F2: NaN/Infinity are not valid numbers for a monetary field assert _datatype_error("monetary_amount", float("nan")) is not None assert _datatype_error("monetary_amount", float("inf")) is not None assert _datatype_error("monetary_amount", 1000) is None # F2/F9: strict ingestion rejects NaN and duplicate keys a lax parser would accept try: load_document('{"a": NaN}'); raise SystemExit("load_document must reject NaN") except ValueError: pass try: load_document('{"a": 1, "a": 2}'); raise SystemExit("load_document must reject dup keys") except ValueError: pass # F3: a Unicode-digit / non-canonical instance suffix -> flagged, never int()-crashes for suffix in ("guarantor:²", "guarantor:٢", "guarantor:2a"): bad = _clone() bad["sections"][suffix] = { "guarantor_name": {"value": "X", "provenance": {"kind": "self-declared"}}} assert any("bad instance suffix" in e for e in validate(bad, catalogue)), suffix # IR5: integer datatype aligns with JSON Schema (accepts 1.0, rejects 1.5 and non-finite) assert _datatype_error("integer", 1.0) is None assert _datatype_error("integer", 1.5) is not None assert _datatype_error("integer", float("nan")) is not None assert _datatype_error("integer", True) is not None # F8: date datatype is anchored + real-calendar (rejects overflow + trailing junk) assert _datatype_error("date", "2026-07-24") is None assert _datatype_error("date", "2026-13-45") is not None assert _datatype_error("date", "2026-02-30") is not None assert _datatype_error("date", "2026-07-24T00:00") is not None # F1/IR7: a malformed overlay selector (no value) does not auto-activate on every pack # (regression guard for the None==None bug) — conventional example stays clean assert not validate(example, catalogue) # ── re-review hardening cases ─────────────────────────────────────────── # #1: a Sharia pack with a non-string sharia_structure value is REJECTED, not crashed bad = json.loads(json.dumps(sharia)) bad["sections"]["sharia_compliance"]["sharia_structure"]["value"] = ["murabaha"] assert validate(bad, catalogue), "list-valued sharia_structure must be rejected, not crash" # #2: a large instance suffix does not int()-crash the validator (returns a list, not raises). bad = _clone() bad["sections"]["guarantor:" + "9" * 5000] = { "guarantor_name": {"value": "X", "provenance": {"kind": "self-declared"}}} assert isinstance(validate(bad, catalogue), list) # no ValueError from int() on 5k digits # a (short) LEADING-ZERO suffix is invalid grammar and is flagged bad = _clone() bad["sections"]["guarantor:0123"] = { "guarantor_name": {"value": "X", "provenance": {"kind": "self-declared"}}} assert any("bad instance suffix" in e for e in validate(bad, catalogue)) # a pathologically huge key/string is refused by the size guard (re-review-3 #9), never crashes bad = _clone() bad["sections"]["guarantor:9" + "9" * 200000] = { "guarantor_name": {"value": "X", "provenance": {"kind": "self-declared"}}} assert any("string longer than" in e for e in validate(bad, catalogue)) # #5: a null value dressed as a verified fact does NOT count toward L2 bad = _clone() for _, fld in _iter_fields(bad): fld["provenance"] = {"kind": "self-declared"} bad["sections"]["business_identity"]["company_name"] = { "value": None, "provenance": {"kind": "verified", "source": "companies_house", "via": "api", "date": "2026-07-10"}} bad["readiness"]["conformance_level"] = "L2-verified-core" bad["readiness"]["status"] = "gaps-remaining" bad["readiness"]["missing"] = ["business_identity.lei"] errs = validate(bad, catalogue) assert any("'verified' requires a substantive value" in e for e in errs), errs assert any("at least one verified fact" in e for e in errs), "null verified must not satisfy L2" # #6: a document over the breadth cap is refused wide = _clone() wide["sections"]["the_ask"].update({ f"x_pad_{i}": {"value": 1, "provenance": {"kind": "self-declared"}} for i in range(_MAX_NODES + 1)}) assert any("breadth guard" in e for e in validate(wide, catalogue)) # #8: a NaN anywhere in the pack is refused up front bad = _clone() bad["sections"]["the_ask"]["amount"]["value"] = float("nan") assert any("non-finite" in e for e in validate(bad, catalogue)) # #4: a stale/wrong catalogue argument is refused even if the pack claims the right version bad_cat = json.loads(json.dumps(catalogue)); bad_cat["catalogue_version"] = "0.9.0" assert any("supplied catalogue" in e for e in validate(example, bad_cat)) # #9: a provided field listed in readiness.missing -> caught bad = _clone() bad["readiness"]["status"] = "gaps-remaining" bad["readiness"]["missing"] = ["business_identity.company_name"] # present + non-null assert any("but it is provided" in e for e in validate(bad, catalogue)) # #14: whitespace attested_by / non-date attested_at -> caught bad = _clone() bad["readiness"]["attestation"] = {"attested_by": " ", "attested_at": "banana"} errs = validate(bad, catalogue) assert any("attested_by" in e for e in errs) and any("attested_at" in e for e in errs), errs # #10/#11/#12: disposition parity — unsupported version, bad instance grammar assert any("disposition_version" in e for e in validate_disposition( {**disp, "disposition_version": "999.0.0"}, catalogue)) bad_d = {**disp, "info_requested": {"field_ids": ["guarantor:01.guarantor_id"]}} assert any("bad instance suffix" in e for e in validate_disposition(bad_d, catalogue)) bad_d = {**disp, "info_requested": {"field_ids": ["the_ask:2.amount"]}} assert any("non-repeatable" in e for e in validate_disposition(bad_d, catalogue)) # #31/#32/#33: stale_facts fails LOUD on bad policy input; future dates are not fresh try: stale_facts(example, "not-a-date"); raise SystemExit("stale_facts must reject bad as_of") except ValueError: pass try: stale_facts(example, "2026-07-14", max_age_days=-1); raise SystemExit("must reject negative age") except ValueError: pass fut = _clone() fut["sections"]["business_identity"]["company_name"]["provenance"]["date"] = "2099-01-01" assert any("company_name" in s for s in stale_facts(fut, "2026-07-14")) # #16/#17: catalogue-integrity — duplicate + reserved-prefix ids are flagged dup_cat = json.loads(json.dumps(catalogue)) dup_cat["sections"].append(dict(dup_cat["sections"][0])) # duplicate a whole section assert any("duplicate section id" in e for e in check_artifacts(dup_cat)) x_cat = json.loads(json.dumps(catalogue)) x_cat["sections"][0]["fields"].append( {"id": "x_reserved", "label": "x", "tag": "declare", "datatype": "string", "definition": "x"}) assert any("reserved x_" in e for e in check_artifacts(x_cat)) # ── re-review-2 (K3) hardening cases ──────────────────────────────────── # H1: a required field with an EMPTY value ("" / []) is NOT provided bad = _clone() bad["sections"]["business_identity"]["company_name"]["value"] = " " assert any("company_name" in e and "not provided" in e for e in validate(bad, catalogue)) # H1: an empty-string verified field is not a verified FACT bad = _clone() for _, fld in _iter_fields(bad): fld["provenance"] = {"kind": "self-declared"} bad["sections"]["business_identity"]["company_name"] = { "value": "", "provenance": {"kind": "verified", "source": "companies_house", "via": "api", "date": "2026-07-10"}} bad["readiness"]["conformance_level"] = "L2-verified-core" bad["readiness"]["status"] = "gaps-remaining" bad["readiness"]["missing"] = ["business_identity.lei"] assert any("at least one verified fact" in e for e in validate(bad, catalogue)) # H2: a catalogue referencing a code_list absent from the artifacts is refused by validate() gutted = json.loads(json.dumps(catalogue)) for s in gutted["sections"]: for f in s["fields"]: f.pop("code_list", None) # keep the catalogue coherent EXCEPT one dangling ref gutted["sections"][0]["fields"][0]["code_list"] = "no_such_list" gutted["sections"][0]["fields"][0]["datatype"] = "enum" assert any("DEGRADED (catalogue)" in e for e in validate(example, gutted)) # H3: an overdrawn BBC (drawn > availability) is caught; headroom must reconcile; # negative amounts are rejected (by the arithmetic rule and/or the schema minimum) assert any("overdrawn" in e for e in validate_bbc(dict(bbc, drawn_balance=999_000_000))) assert any("reconcile" in e for e in validate_bbc(dict(bbc, headroom=1))) assert validate_bbc(dict(bbc, drawn_balance=-5)), "a negative amount must be rejected" # H4: an expired verification is stale even if recently dated exp = _clone() exp["sections"]["business_identity"]["company_name"]["provenance"]["expires_at"] = "2020-01-01" assert any("company_name" in s for s in stale_facts(exp, "2026-07-14")) # M2: a disposition with a non-date timestamp is caught assert any("timestamp" in e for e in validate_disposition({**disp, "timestamp": "banana"}, catalogue)) # M4: a bare unknown section (empty, or x_-only) is rejected bad = _clone(); bad["sections"]["evil_corp"] = {} assert any("unknown section" in e for e in validate(bad, catalogue)) bad = _clone(); bad["sections"]["evil_corp"] = { "x_a": {"value": 1, "provenance": {"kind": "self-declared"}}} assert any("unknown section" in e for e in validate(bad, catalogue)) # M5: a catalogue PATCH (0.10.0 -> 0.10.9) does NOT invalidate a 0.10.0 pack patched = json.loads(json.dumps(catalogue)); patched["catalogue_version"] = "0.11.9" assert not validate(example, patched), validate(example, patched) # M6: attestation object present but attested not true -> contradictory bad = _clone(); bad["readiness"]["attested"] = False assert any("contradictory signals" in e for e in validate(bad, catalogue)) # L1: load_document raises ValueError (not RecursionError) on pathologically deep input deep_txt = "[" * 5000 + "]" * 5000 try: load_document(deep_txt) except ValueError: pass except RecursionError: raise SystemExit("load_document must raise ValueError, not RecursionError") # L5: a non-numeric FRN is flagged bad = _clone(); bad["producer"] = {"name": "X Ltd", "fca_status": "authorised", "fca_frn": "notanfrn"} assert any("6" in e and "digit" in e for e in validate(bad, catalogue)) # ── re-review-3 (convergence) interaction cases ───────────────────────── # C1: load_document rejects 1e999 (overflow to inf), not just literal NaN/Infinity for bad_num in ('{"a": 1e999}', '{"a": -1e999}', '{"a": NaN}'): try: load_document(bad_num); raise SystemExit(f"load_document must reject {bad_num}") except ValueError: pass # C2: load_document + validate reject a lone surrogate try: load_document('{"a": "\\ud800"}'); raise SystemExit("load_document must reject lone surrogate") except ValueError: pass bad = _clone(); bad["sections"]["business_identity"]["company_name"]["value"] = "x\ud800y" assert any("surrogate" in e for e in validate(bad, catalogue)) # C3: a value:"" field listed in readiness.missing is VALID (no trap) — H1 and #9 agree ok = _clone() ok["sections"]["business_identity"]["company_name"] = { "value": "", "provenance": {"kind": "self-declared"}} ok["readiness"]["status"] = "gaps-remaining" ok["readiness"]["missing"] = ["business_identity.company_name"] errs = validate(ok, catalogue) assert not any("but it is provided" in e for e in errs), errs # the contradiction is gone # C4/C6: a malformed catalogue is REFUSED cleanly, never crashes (the layer whose job it is) for badcat in ({"catalogue_version": "0.10.0", "sections": ["oops"]}, {"catalogue_version": "0.10.0", "sections": [{"id": 5, "title": "x", "fields": []}]}, {"catalogue_version": "0.10.0", "sections": [{"id": None, "title": "x", "fields": [{}]}]}): r = validate(example, badcat) assert isinstance(r, list) and r, f"malformed catalogue must be refused: {badcat}" assert isinstance(check_artifacts({"sections": ["x"]}), list) # direct call: no crash # H1: an overlay selector typo is caught by the consistency battery ov_cat = json.loads(json.dumps(catalogue)) # (validate the real overlays are clean first — already asserted via check_artifacts above) assert not any("overlay" in e for e in check_artifacts(catalogue)), "real overlays must be clean" # H3: a disposition still validates when the profiles artifact is irrelevant (not coupled) assert not validate_disposition(disp, catalogue), validate_disposition(disp, catalogue) # H4: stale_facts does not crash on a malformed pack assert stale_facts({"sections": {"s": "notadict"}}, "2026-07-14") == [] assert stale_facts({"sections": {"s": {"f": "notadict"}}}, "2026-07-14") == [] # M3: an x_ field with a non-conforming token is flagged bad = _clone() bad["sections"]["the_ask"]["x_Bad"] = {"value": 1, "provenance": {"kind": "self-declared"}} assert any("x_[a-z]" in e for e in validate(bad, catalogue)) # H5: BBC ineligible reconciliation + period ordering assert any("reconcile" in e for e in validate_bbc(dict(bbc, eligible_receivables=449999))) assert any("period_end is before" in e for e in validate_bbc( dict(bbc, period_start="2026-06-30", period_end="2026-06-01"))) # #26: a valid disposition status raises no disposition_status code-list error (the # code-list check is defence-in-depth; the schema enum blocks a truly invalid status first) assert not any("disposition_status" in e for e in validate_disposition(disp, catalogue)) # ── fresh-eyes (3rd-model) review cases ──────────────────────────────── # V-01: an INVISIBLE character must not sneak an excluded activity past the sector # gate. Each of these renders as "92000" to a human but dodged a "92" prefix match. for sneaky in ("​92000", "9‌2000", "92­000", "‎92000"): bad = json.loads(json.dumps(sharia)) bad["sections"]["business_identity"]["sic_codes"] = { "value": [sneaky], "provenance": {"kind": "verified", "source": "companies_house", "via": "api", "date": "2026-07-10"}} errs = validate(bad, catalogue) assert any("sector gate" in e for e in errs), f"invisible-char SIC {sneaky!r} bypassed: {errs}" # a nested / non-string code cannot be evaluated -> refused, not passed bad = json.loads(json.dumps(sharia)) bad["sections"]["business_identity"]["sic_codes"] = { "value": [["92000"]], "provenance": {"kind": "verified", "source": "companies_house", "via": "api", "date": "2026-07-10"}} assert any("sector gate" in e for e in validate(bad, catalogue)) # and a legitimately-formatted excluded code is still caught, separators and all for ok_form in ("92000", " 92000 ", "92-000"): bad = json.loads(json.dumps(sharia)) bad["sections"]["business_identity"]["sic_codes"] = { "value": [ok_form], "provenance": {"kind": "verified", "source": "companies_house", "via": "api", "date": "2026-07-10"}} assert any("matches an excluded prefix" in e for e in validate(bad, catalogue)), ok_form # V-06: a date the ordering gate cannot parse is refused, not silently skipped bad = json.loads(json.dumps(sharia)) bad["sections"]["sharia_compliance"]["ownership_transfer_date"]["value"] = "2026-07-10T00:00:00" assert any("cannot be evaluated" in e for e in validate(bad, catalogue)) # V-10: a malformed FRN is checked even when no fca_status demands one bad = _clone() bad["producer"] = {"name": "X Ltd", "fca_frn": "notanfrn"} assert any("Firm Reference Number" in e for e in validate(bad, catalogue)) # ── ABL: a borrowing base carrying TWO asset classes ──────────────────── abl = dict(bbc, gross_inventory=300000.0, ineligible_inventory={"aged": 50000.0, "wip": 25000.0}, eligible_inventory=225000.0, inventory_nolv_rate=0.6, inventory_advance_rate=0.35, inventory_availability=78750.0, availability=461250.0, headroom=161250.0) assert not validate_bbc(abl), validate_bbc(abl) # the single-line receivables rule must NOT fire on a multi-asset base (it would on # every ABL certificate), but must still bite on a receivables-only one assert any("eligible_receivables x advance_rate" in e for e in validate_bbc(dict(bbc, availability=999999.0))) assert any("exceeds gross_inventory" in e for e in validate_bbc(dict(abl, eligible_inventory=400000.0))) assert any("does not reconcile" in e for e in validate_bbc(dict(abl, eligible_inventory=250000.0))) assert any("inventory_availability exceeds" in e for e in validate_bbc(dict(abl, inventory_availability=200000.0, availability=582500.0, headroom=282500.0))) assert any("plus the inventory line" in e for e in validate_bbc(dict(abl, availability=600000.0, headroom=300000.0))) # an inventory advance rate with no appraised NOLV basis is named assert any("NOLV" in e for e in validate_bbc( {k: v for k, v in abl.items() if k != "inventory_nolv_rate"})) # ── money is DECIMAL, not binary float (v0.6: sec review #27/#34) ──────── _b = {k: v for k, v in bbc.items() if k not in ("gross_receivables", "ineligible", "eligible_receivables", "advance_rate", "availability", "drawn_balance", "headroom")} # the classic float failure: 0.1 + 0.2 != 0.3 in binary. It must reconcile here. assert not validate_bbc(dict(_b, gross_receivables=0.3, ineligible={"aged_over_limit": 0.1, "contra": 0.2}, eligible_receivables=0.0, advance_rate=0.85, availability=0.0, drawn_balance=0.0, headroom=0.0)) # cent precision at scale — float64 silently loses pennies on a £100m facility _big = dict(_b, gross_receivables=100_000_000.01, ineligible={"aged_over_limit": 0.01}, eligible_receivables=100_000_000.00, advance_rate=1.0, availability=100_000_000.00, drawn_balance=0.0, headroom=100_000_000.00) assert not validate_bbc(_big), validate_bbc(_big) # ...and being a single penny out at that scale is still caught assert any("reconcile" in e for e in validate_bbc( dict(_big, eligible_receivables=100_000_000.01, availability=100_000_000.01, headroom=100_000_000.01))) # the old `+ 0.01` tolerance silently accepted EXACTLY one penny of overstatement # (the comparison was `>`, not `>=`). It is gone. assert any("advance_rate" in e for e in validate_bbc( dict(bbc, availability=382500.01, headroom=82500.01))) # ── catalogue digest pinning (v0.6: sec review M8/#1) ─────────────────── # The catalogue is a TRUST ROOT: it decides which fields exist, which are required, and # which are verify-tagged. Until now the only check was its own self-declared version # string, so an attacker-authored — or merely stale — catalogue that claimed 0.11.x and # was internally coherent would be accepted and could bless a gutted pack. # Enforcement is OPT-IN, consistent with how signature keys and evidence responses work: # the validator gives the caller the means to pin, it cannot know which catalogue the # caller trusts. _dg = catalogue_digest(catalogue) assert _dg.startswith("sha256:") and len(_dg) == 71 # stable across re-serialization (order/whitespace must not change the digest) assert catalogue_digest(json.loads(json.dumps(catalogue))) == _dg # pinning to the right digest is transparent assert not validate(example, catalogue, expect_catalogue_digest=_dg) # pinning to a WRONG digest refuses, even though the catalogue is internally coherent assert any("catalogue digest" in e for e in validate(example, catalogue, expect_catalogue_digest="sha256:" + "0" * 64)) # a tampered-but-coherent catalogue changes the digest — which is the whole point _tampered = json.loads(json.dumps(catalogue)) _tampered["sections"][0]["fields"][0]["definition"] = "quietly altered" assert catalogue_digest(_tampered) != _dg assert any("catalogue digest" in e for e in validate(example, _tampered, expect_catalogue_digest=_dg)) # ── Unicode normalization (v0.6: sec review #50) ──────────────────────── # "Café" has two byte representations that render identically: NFC (é = U+00E9) and # NFD (e + combining acute U+0301). RFC 8785 deliberately does NOT normalise, so the # two forms produce DIFFERENT canonical bytes — hence a different pack_ref and a # different evidence_hash for a pack a human would call identical. macOS filesystems # hand back NFD, so this is a real cross-producer hazard, not a theoretical one. import unicodedata as _ud _nfc = "Café Trading Ltd" _nfd = _ud.normalize("NFD", _nfc) assert _nfc != _nfd and _ud.normalize("NFC", _nfd) == _nfc # they really do differ # a pack carrying non-NFC text must be FLAGGED — its identity is unstable elsewhere bad = _clone() bad["sections"]["business_identity"]["company_name"]["value"] = _nfd assert any("NFC" in e for e in validate(bad, catalogue)), \ f"non-NFC text must be flagged: {validate(bad, catalogue)[:2]}" # ...and the NFC form of the same name is clean (no false positive) ok = _clone() ok["sections"]["business_identity"]["company_name"]["value"] = _nfc assert not validate(ok, catalogue), validate(ok, catalogue) # a non-NFC SECTION/FIELD KEY is caught too — keys are hashed just like values bad = _clone() bad["sections"]["business_identity"][_ud.normalize("NFD", "x_café")] = { "value": 1, "provenance": {"kind": "self-declared"}} assert any("NFC" in e for e in validate(bad, catalogue)) # ── required fields apply PER INSTANCE (v0.6: sec review M7/#15) ───────── # A second guarantor carrying one harmless field used to leave the pack oven-ready. _orig_rff = required_fields_for try: # temporarily require a guarantor field, which no shipped profile does today globals()["required_fields_for"] = lambda p, c: ( list(_orig_rff(p, c)[0]) + ["guarantor.guarantor_id"], _orig_rff(p, c)[1]) _inst = json.loads(json.dumps(multi)) _inst["sections"]["guarantor:2"].pop("guarantor_id", None) _inst["readiness"]["missing"] = [m for m in _inst["readiness"]["missing"] if "guarantor:2" not in m] _e = check_required(_inst, catalogue) assert any("guarantor:2.guarantor_id" in x for x in _e), _e # ...and declaring the instance gap honestly clears it _inst["readiness"]["missing"].append("guarantor:2.guarantor_id") assert not any("guarantor:2" in x for x in check_required(_inst, catalogue)) finally: globals()["required_fields_for"] = _orig_rff print("deal-package standard self-check OK — ready + multi examples valid (reference AND " "strict schema); artifacts cross-validate; 34 baseline + 3 rounds of security-review " "hardening cases caught (invented field, " "incoherent readiness, self-declared-L3, entity gate, empty attestation x2, wrong " "source, tag mismatch, unmet required field, non-prefixed unknown field, bad " "code-list value, per-product profile gaps x2, non-repeatable instance, bad/zero-pad " "instance suffix x2, orphan instance, phantom missing-instance, datatype x2, " "disposition x4) + x_ extension + taxonomy-only fallback + freshness helper; PLUS " "3 review rounds: R1 (IR1 forgery, IR2/IR3 version+crash, F2 NaN, F3 unicode, F8 date), " "R2 (sharia/suffix crashes, verified-null, disposition/BBC parity, empty-value forgery, " "expires_at, breadth cap), R3 (fail-closed wrapper, 1e999/surrogate ingestion, " "value:'' contradiction, malformed-catalogue refusal, overlay battery, BBC reconcile)")