#!/usr/bin/env python3 """Validator for the CEDE canonical risk object schema (SPEC.md section 2). Standard library only, by design: this file ships in the public schema repo (SPEC.md section 5) and every harness script in this repo runs on a bare python3, so consuming the schema must never require a package install. What it does, in order: 1. lints the schema itself (`--check-schema`): every keyword must be one this validator implements, with a value of the right shape, and every $ref must resolve. An unknown keyword is a hard error, never a silent skip — a validator that ignores keywords it does not understand under-validates without saying so; 2. validates instances against the schema; 3. applies the SPEC constraints JSON Schema cannot express (below); 4. runs the committed example suite against schema/examples/expectations.json, which pins not only that each invalid example is rejected but *why* — a validator that rejected everything would fail the valid examples, and one that accepted everything would fail the invalid ones. Semantic rules (JSON Schema cannot compare two sibling instance values, nor consult the tz database): S1 period.expiry must be strictly after period.inception (SPEC section 2, period) S2 period.timezone must name a real IANA zone (SPEC section 2, period) Cross-check: when the `jsonschema` package happens to be importable, every verdict is computed twice — once here, once by that reference implementation — and any disagreement is a failure. When it is absent (bare CI), that is reported loudly rather than passed over, and the schema lint stands alone. Usage: python3 schema/validate.py run the committed example suite python3 schema/validate.py --check-schema lint the schema only python3 schema/validate.py FILE [FILE ...] validate instance files python3 schema/validate.py --schema PATH ... use an alternate schema python3 schema/validate.py --quiet suite output, failures only Exit: 0 when everything asked for passed, nonzero otherwise. """ from __future__ import annotations import argparse import datetime as _dt import json import os import re import sys HERE = os.path.dirname(os.path.abspath(__file__)) DEFAULT_SCHEMA = os.path.join(HERE, "risk-object.v0.schema.json") DEFAULT_EXPECTATIONS = os.path.join(HERE, "examples", "expectations.json") # -------------------------------------------------------------------------- # the keyword set this validator implements # -------------------------------------------------------------------------- # Assertions: each one is applied. Adding a keyword here means implementing it # in _validate below; the lint refuses any keyword absent from both sets, so # the two can never drift apart silently. ASSERTIONS = { "$ref", "type", "enum", "const", "required", "properties", "additionalProperties", "minProperties", "maxProperties", "items", "minItems", "maxItems", "uniqueItems", "minLength", "maxLength", "pattern", "minimum", "maximum", "exclusiveMinimum", "exclusiveMaximum", "multipleOf", "allOf", "anyOf", "oneOf", "not", "if", "then", "else", } # Annotations: carried for readers, asserting nothing. ANNOTATIONS = { "$schema", "$id", "$defs", "$comment", "title", "description", "examples", "default", "deprecated", } KNOWN = ASSERTIONS | ANNOTATIONS # Keyword -> the JSON types its value may take, checked by the lint. _SCHEMA = "schema" # a subschema (object or boolean) _SCHEMA_LIST = "schemalist" # a non-empty array of subschemas _SCHEMA_MAP = "schemamap" # an object whose values are subschemas KEYWORD_VALUE = { "$ref": str, "$schema": str, "$id": str, "$comment": str, "title": str, "description": str, "enum": list, "required": list, "minProperties": int, "maxProperties": int, "minItems": int, "maxItems": int, "uniqueItems": bool, "minLength": int, "maxLength": int, "pattern": str, "minimum": (int, float), "maximum": (int, float), "exclusiveMinimum": (int, float), "exclusiveMaximum": (int, float), "multipleOf": (int, float), "properties": _SCHEMA_MAP, "$defs": _SCHEMA_MAP, "items": _SCHEMA, "additionalProperties": _SCHEMA, "not": _SCHEMA, "if": _SCHEMA, "then": _SCHEMA, "else": _SCHEMA, "allOf": _SCHEMA_LIST, "anyOf": _SCHEMA_LIST, "oneOf": _SCHEMA_LIST, # "type", "const", "examples", "default", "deprecated" are shape-checked # individually below. } JSON_TYPES = {"object", "array", "string", "number", "integer", "boolean", "null"} class SchemaError(Exception): """The schema itself is malformed or uses an unsupported keyword.""" # -------------------------------------------------------------------------- # instance type inspection # -------------------------------------------------------------------------- def json_type(value): if value is None: return "null" if isinstance(value, bool): # before int: bool is an int subclass return "boolean" if isinstance(value, str): return "string" if isinstance(value, int): return "integer" if isinstance(value, float): return "integer" if value.is_integer() else "number" if isinstance(value, list): return "array" if isinstance(value, dict): return "object" raise SchemaError("value is not JSON data: %r" % (value,)) def type_matches(value, wanted): actual = json_type(value) if wanted == "number": return actual in ("number", "integer") return actual == wanted def canonical(value): return json.dumps(value, sort_keys=True, separators=(",", ":")) def equal(a, b): # JSON equality: 1 and 1.0 are the same number, True and 1 are not. if isinstance(a, bool) != isinstance(b, bool): return False if isinstance(a, (int, float)) and isinstance(b, (int, float)): return a == b if type(a) is not type(b): return False return canonical(a) == canonical(b) # -------------------------------------------------------------------------- # schema lint # -------------------------------------------------------------------------- def resolve_ref(root, ref, where): if not ref.startswith("#"): raise SchemaError( "%s: $ref %r is not a local reference — this validator resolves " "same-document JSON pointers only, so the schema stays consumable " "with no network access" % (where, ref)) pointer = ref[1:] if pointer and not pointer.startswith("/"): raise SchemaError("%s: $ref %r is not a JSON pointer" % (where, ref)) target = root for raw in [p for p in pointer.split("/") if p != ""]: token = raw.replace("~1", "/").replace("~0", "~") if not isinstance(target, dict) or token not in target: raise SchemaError("%s: $ref %r does not resolve" % (where, ref)) target = target[token] if not isinstance(target, (dict, bool)): raise SchemaError("%s: $ref %r resolves to something that is not a schema" % (where, ref)) return target def lint_schema(schema, root=None, where="#"): """Raise SchemaError unless every keyword is implemented and well formed.""" if root is None: root = schema if not isinstance(schema, dict): raise SchemaError("the root schema must be an object") declared = schema.get("$schema") if declared != "https://json-schema.org/draft/2020-12/schema": raise SchemaError( "root $schema must declare draft 2020-12 " "(https://json-schema.org/draft/2020-12/schema), got %r" % (declared,)) if isinstance(schema, bool): return if not isinstance(schema, dict): raise SchemaError("%s: subschema must be an object or a boolean" % where) for key, value in schema.items(): spot = "%s/%s" % (where, key) if key not in KNOWN: raise SchemaError( "%s: unsupported keyword %r. This validator refuses to load a " "schema whose keywords it cannot enforce — silently ignoring a " "keyword would under-validate every object" % (spot, key)) expect = KEYWORD_VALUE.get(key) if expect is _SCHEMA: lint_schema(value, root, spot) elif expect is _SCHEMA_LIST: if not isinstance(value, list) or not value: raise SchemaError("%s: must be a non-empty array of schemas" % spot) for i, sub in enumerate(value): lint_schema(sub, root, "%s/%d" % (spot, i)) elif expect is _SCHEMA_MAP: if not isinstance(value, dict): raise SchemaError("%s: must be an object of subschemas" % spot) for name, sub in value.items(): lint_schema(sub, root, "%s/%s" % (spot, name)) elif expect is not None: if isinstance(expect, type) and expect is int: ok = isinstance(value, int) and not isinstance(value, bool) elif isinstance(expect, type) and expect is bool: ok = isinstance(value, bool) elif expect == (int, float): ok = isinstance(value, (int, float)) and not isinstance(value, bool) else: ok = isinstance(value, expect) if not ok: raise SchemaError("%s: value has the wrong type: %r" % (spot, value)) if key == "type": names = value if isinstance(value, list) else [value] if not names or not all(n in JSON_TYPES for n in names): raise SchemaError("%s: not a JSON type name (or list of them): %r" % (spot, value)) elif key == "enum": if not value: raise SchemaError("%s: enum must list at least one value" % spot) if len({canonical(v) for v in value}) != len(value): raise SchemaError("%s: enum has duplicate values" % spot) elif key == "required": if not all(isinstance(n, str) for n in value): raise SchemaError("%s: required must list property names" % spot) if len(set(value)) != len(value): raise SchemaError("%s: required names a property twice" % spot) elif key == "pattern": try: re.compile(value) except re.error as exc: raise SchemaError("%s: pattern does not compile: %s" % (spot, exc)) elif key == "$ref": resolve_ref(root, value, spot) if "then" in schema and "if" not in schema: raise SchemaError("%s: 'then' without 'if' asserts nothing" % where) if "else" in schema and "if" not in schema: raise SchemaError("%s: 'else' without 'if' asserts nothing" % where) return True # -------------------------------------------------------------------------- # instance validation # -------------------------------------------------------------------------- def _err(errors, path, message): errors.append("%s: %s" % (path or "/", message)) def _validate(instance, schema, root, path, errors): if schema is True: return if schema is False: _err(errors, path, "no value is permitted here") return if "$ref" in schema: _validate(instance, resolve_ref(root, schema["$ref"], path), root, path, errors) if "type" in schema: wanted = schema["type"] names = wanted if isinstance(wanted, list) else [wanted] if not any(type_matches(instance, n) for n in names): _err(errors, path, "type: expected %s, got %s" % (" or ".join(names), json_type(instance))) return # further assertions on the wrong type only add noise if "enum" in schema and not any(equal(instance, v) for v in schema["enum"]): _err(errors, path, "enum: %s is not one of %s" % (canonical(instance), canonical(schema["enum"]))) if "const" in schema and not equal(instance, schema["const"]): _err(errors, path, "const: expected %s, got %s" % (canonical(schema["const"]), canonical(instance))) kind = json_type(instance) if kind == "object": for name in schema.get("required", []): if name not in instance: _err(errors, path, "required: property %r is missing" % name) props = schema.get("properties", {}) for name, value in instance.items(): if name in props: _validate(value, props[name], root, "%s/%s" % (path, name), errors) if "additionalProperties" in schema: extra = schema["additionalProperties"] for name in [n for n in instance if n not in props]: if extra is False: _err(errors, "%s/%s" % (path, name), "additionalProperties: property %r is not permitted here" % name) else: _validate(instance[name], extra, root, "%s/%s" % (path, name), errors) if "minProperties" in schema and len(instance) < schema["minProperties"]: _err(errors, path, "minProperties: %d propertie(s), need at least %d" % (len(instance), schema["minProperties"])) if "maxProperties" in schema and len(instance) > schema["maxProperties"]: _err(errors, path, "maxProperties: %d propertie(s), permitted at most %d" % (len(instance), schema["maxProperties"])) elif kind == "array": if "items" in schema: for i, item in enumerate(instance): _validate(item, schema["items"], root, "%s/%d" % (path, i), errors) if "minItems" in schema and len(instance) < schema["minItems"]: _err(errors, path, "minItems: %d item(s), need at least %d" % (len(instance), schema["minItems"])) if "maxItems" in schema and len(instance) > schema["maxItems"]: _err(errors, path, "maxItems: %d item(s), permitted at most %d" % (len(instance), schema["maxItems"])) if schema.get("uniqueItems") and len({canonical(i) for i in instance}) != len(instance): _err(errors, path, "uniqueItems: the array repeats a value") elif kind == "string": if "minLength" in schema and len(instance) < schema["minLength"]: _err(errors, path, "minLength: %d character(s), need at least %d" % (len(instance), schema["minLength"])) if "maxLength" in schema and len(instance) > schema["maxLength"]: _err(errors, path, "maxLength: %d character(s), permitted at most %d" % (len(instance), schema["maxLength"])) if "pattern" in schema and re.search(schema["pattern"], instance) is None: _err(errors, path, "pattern: %r does not match %s" % (instance, schema["pattern"])) elif kind in ("number", "integer"): if "minimum" in schema and instance < schema["minimum"]: _err(errors, path, "minimum: %s is below %s" % (instance, schema["minimum"])) if "maximum" in schema and instance > schema["maximum"]: _err(errors, path, "maximum: %s is above %s" % (instance, schema["maximum"])) if "exclusiveMinimum" in schema and instance <= schema["exclusiveMinimum"]: _err(errors, path, "exclusiveMinimum: %s is not above %s" % (instance, schema["exclusiveMinimum"])) if "exclusiveMaximum" in schema and instance >= schema["exclusiveMaximum"]: _err(errors, path, "exclusiveMaximum: %s is not below %s" % (instance, schema["exclusiveMaximum"])) if "multipleOf" in schema: step = schema["multipleOf"] if step <= 0 or (instance / step) != int(instance / step): _err(errors, path, "multipleOf: %s is not a multiple of %s" % (instance, step)) for sub in schema.get("allOf", []): _validate(instance, sub, root, path, errors) if "anyOf" in schema: if not any(passes(instance, sub, root) for sub in schema["anyOf"]): _err(errors, path, "anyOf: matches none of the %d permitted forms" % len(schema["anyOf"])) if "oneOf" in schema: hits = [i for i, sub in enumerate(schema["oneOf"]) if passes(instance, sub, root)] if len(hits) != 1: titles = [s.get("title", "form %d" % i) if isinstance(s, dict) else "form %d" % i for i, s in enumerate(schema["oneOf"])] _err(errors, path, "oneOf: matches %d of the %d permitted forms (%s), " "exactly one is required" % (len(hits), len(schema["oneOf"]), ", ".join(titles))) if "not" in schema and passes(instance, schema["not"], root): _err(errors, path, "not: this value is excluded here") if "if" in schema: branch = "then" if passes(instance, schema["if"], root) else "else" if branch in schema: _validate(instance, schema[branch], root, path, errors) def passes(instance, schema, root): errors = [] _validate(instance, schema, root, "", errors) return not errors def validate_instance(instance, schema): """Return a list of human-readable error strings (empty means valid).""" errors = [] _validate(instance, schema, schema, "", errors) return errors # -------------------------------------------------------------------------- # SPEC constraints JSON Schema cannot express # -------------------------------------------------------------------------- def _parse_timestamp(text): try: return _dt.datetime.fromisoformat(text.replace("Z", "+00:00")) except (ValueError, AttributeError): return None def semantic_errors(instance): """SPEC section 2 rules that no JSON Schema keyword can state.""" errors = [] if not isinstance(instance, dict): return errors period = instance.get("period") if isinstance(period, dict): start = _parse_timestamp(period.get("inception")) end = _parse_timestamp(period.get("expiry")) if start is not None and end is not None and end <= start: _err(errors, "/period/expiry", "S1: expiry (%s) must be after inception (%s) — SPEC section 2, period" % (period.get("expiry"), period.get("inception"))) zone = period.get("timezone") if isinstance(zone, str): known = _available_timezones() if known is None: pass # reported once, loudly, by the caller elif zone not in known: _err(errors, "/period/timezone", "S2: %r is not a name in the IANA tz database" % zone) return errors _TZ_CACHE = [] def _available_timezones(): """The IANA zone names on this machine, or None if there is no tz database.""" if not _TZ_CACHE: try: import zoneinfo names = set(zoneinfo.available_timezones()) except Exception: names = None _TZ_CACHE.append(names or None) return _TZ_CACHE[0] # -------------------------------------------------------------------------- # reference cross-check # -------------------------------------------------------------------------- def _reference_validator(schema): """Draft 2020-12 validation from the `jsonschema` package, when installed.""" try: import jsonschema except ImportError: return None try: cls = jsonschema.Draft202012Validator except AttributeError: return None cls.check_schema(schema) return cls(schema) def cross_check(instance, schema, mine, reference): """Fail loudly if this validator and the reference disagree on a verdict.""" if reference is None: return [] theirs = list(reference.iter_errors(instance)) if bool(mine) == bool(theirs): return [] return ["CROSS-CHECK DISAGREEMENT: schema/validate.py says %s, the jsonschema " "reference implementation says %s (%s)" % ("invalid" if mine else "valid", "invalid" if theirs else "valid", "; ".join(e.message for e in theirs) or "; ".join(mine))] # -------------------------------------------------------------------------- # runners # -------------------------------------------------------------------------- def load_json(path): with open(path, encoding="utf-8") as handle: return json.load(handle) def check_object(instance, schema, reference): """Full verdict for one instance: schema errors, then semantic errors.""" schema_errors = validate_instance(instance, schema) # The reference implementation knows nothing of the semantic rules, so it # is compared against the schema verdict alone. return (schema_errors + semantic_errors(instance) + cross_check(instance, schema, schema_errors, reference)) def run_suite(schema_path, expectations_path, quiet=False): schema = load_json(schema_path) lint_schema(schema) reference = _reference_validator(schema) print("schema: %s" % os.path.relpath(schema_path, HERE)) print("schema lint: OK — every keyword implemented, every $ref resolves") if reference is None: print(">>> NOTE: the `jsonschema` package is not importable here, so the " "schema was checked against this validator's draft 2020-12 keyword " "lint only, not against the published metaschema, and no " "cross-check ran. Install it to get both.") else: print("metaschema: OK — valid draft 2020-12 per the jsonschema reference " "implementation; every verdict below is cross-checked against it") if _available_timezones() is None: print(">>> NOTE: no IANA tz database on this machine — semantic rule S2 " "(period.timezone names a real zone) could not run.") expectations = load_json(expectations_path) base = os.path.dirname(expectations_path) failures = 0 counts = {"valid": 0, "invalid": 0} for case in expectations["cases"]: path = os.path.join(base, case["file"]) expect_valid = case["expect"] == "valid" try: instance = load_json(path) except (OSError, ValueError) as exc: print("FAIL %-46s cannot be read: %s" % (case["file"], exc)) failures += 1 continue errors = check_object(instance, schema, reference) joined = " | ".join(errors) problems = [] if expect_valid and errors: problems.append("expected valid, rejected: %s" % joined) elif not expect_valid and not errors: problems.append("expected invalid, accepted") elif not expect_valid: needle = case.get("expect_error_contains") if needle and needle not in joined: problems.append("rejected for the wrong reason: wanted %r, got: %s" % (needle, joined)) if any(e.startswith("CROSS-CHECK") for e in errors): problems.append(joined) if problems: failures += 1 print("FAIL %-46s %s" % (case["file"], "; ".join(problems))) else: counts["valid" if expect_valid else "invalid"] += 1 if not quiet: print("PASS %-46s %s" % ( case["file"], case["why"] if expect_valid else "rejected: %s" % (case.get("expect_error_contains") or "as expected"))) print("-" * 78) print("examples: %d valid accepted, %d invalid rejected, %d failure(s)" % (counts["valid"], counts["invalid"], failures)) if failures: print("SCHEMA EXAMPLE SUITE: RED") return 1 if counts["valid"] < 2 or counts["invalid"] < 2: print("SCHEMA EXAMPLE SUITE: RED — the suite must hold at least two valid " "and two invalid examples; it holds %d and %d" % (counts["valid"], counts["invalid"])) return 1 print("SCHEMA EXAMPLE SUITE: GREEN") return 0 def _walk(value, path=()): yield path, value if isinstance(value, dict): for key, sub in value.items(): yield from _walk(sub, path + (key,)) elif isinstance(value, list): for i, sub in enumerate(value): yield from _walk(sub, path + (i,)) def _mutate(document, path, kind): """Return a copy of document with one edit applied, or None if not applicable.""" import copy as _copy mutant = _copy.deepcopy(document) parent = mutant for step in path[:-1]: parent = parent[step] last = path[-1] if kind == "delete": del parent[last] elif kind == "duplicate-item": target = parent[last] if not isinstance(target, list) or not target: return None parent[last] = target + [target[0]] else: parent[last] = kind return mutant def run_self_test(schema_path, expectations_path): """Differential test: mutate every node of every valid example in several ways and require this validator and the reference implementation to reach the same verdict on every mutant. The example suite proves the validator is not blind at ten points; this sweeps thousands.""" schema = load_json(schema_path) lint_schema(schema) reference = _reference_validator(schema) if reference is None: print(">>> SELF-TEST SKIPPED: the differential sweep needs the " "`jsonschema` package to disagree with, and it is not importable " "here. This is a skip, not a pass — install it and rerun.") return 0 expectations = load_json(expectations_path) base = os.path.dirname(expectations_path) edits = ["delete", "duplicate-item", "x", 7, -3, None, {}, [], True] mutants = disagreements = 0 for case in expectations["cases"]: if case["expect"] != "valid": continue document = load_json(os.path.join(base, case["file"])) for path, _ in list(_walk(document)): if not path: continue for kind in edits: try: mutant = _mutate(document, path, kind) except (KeyError, IndexError, TypeError): continue if mutant is None: continue mutants += 1 mine = bool(validate_instance(mutant, schema)) theirs = bool(list(reference.iter_errors(mutant))) if mine != theirs: disagreements += 1 if disagreements <= 10: print("SELF-TEST FAIL: %s at /%s (%r): this validator says " "%s, the reference says %s" % (case["file"], "/".join(str(p) for p in path), kind, "invalid" if mine else "valid", "invalid" if theirs else "valid")) print("self-test: %d mutant(s) of the valid examples, %d disagreement(s) " "with the jsonschema reference implementation" % (mutants, disagreements)) if disagreements: print("VALIDATOR SELF-TEST: RED") return 1 if mutants < 100: print("VALIDATOR SELF-TEST: RED — too few mutants (%d) for the sweep to " "mean anything" % mutants) return 1 print("VALIDATOR SELF-TEST: OK") return 0 def run_files(schema_path, paths): schema = load_json(schema_path) lint_schema(schema) reference = _reference_validator(schema) bad = 0 for path in paths: try: instance = load_json(path) except (OSError, ValueError) as exc: print("INVALID %s — cannot be read: %s" % (path, exc)) bad += 1 continue errors = check_object(instance, schema, reference) if errors: bad += 1 print("INVALID %s" % path) for message in errors: print(" %s" % message) else: print("VALID %s" % path) return 1 if bad else 0 def main(argv=None): parser = argparse.ArgumentParser( description="Validate risk objects against the CEDE v0 schema.") parser.add_argument("files", nargs="*", help="instance files to validate; none runs the example suite") parser.add_argument("--schema", default=DEFAULT_SCHEMA) parser.add_argument("--expectations", default=DEFAULT_EXPECTATIONS) parser.add_argument("--check-schema", action="store_true", help="lint the schema and exit") parser.add_argument("--self-test", action="store_true", help="differential sweep against the jsonschema reference") parser.add_argument("--quiet", action="store_true", help="in the example suite, print failures only") args = parser.parse_args(argv) try: if args.check_schema: schema = load_json(args.schema) lint_schema(schema) print("schema lint: OK — %s" % args.schema) reference = _reference_validator(schema) if reference is None: print(">>> NOTE: `jsonschema` is not importable, so the published " "draft 2020-12 metaschema was not consulted.") else: print("metaschema: OK — valid draft 2020-12 per the jsonschema " "reference implementation") return 0 if args.self_test: return run_self_test(args.schema, args.expectations) if args.files: return run_files(args.schema, args.files) return run_suite(args.schema, args.expectations, quiet=args.quiet) except SchemaError as exc: print("SCHEMA ERROR: %s" % exc) return 2 if __name__ == "__main__": sys.exit(main())