#!/usr/bin/env python3 """Offline educational fixture verifier, Python 3.10+. Install dependency separately: python -m pip install jsonschema[format]==4.26.0 Keep this file beside trading-webhook-record-only.{schema,example,fixtures}.json. Run: python verify-trading-webhook-offline.py Reads those three bounded local files only. No HTTP client, receiver, credentials, queue, database or order submission. This is not a TradeCopier API integration. """ import hashlib import json import sys from datetime import datetime, timedelta from decimal import Decimal, InvalidOperation from pathlib import Path from jsonschema import Draft202012Validator, FormatChecker from jsonschema.exceptions import SchemaError, ValidationError from referencing import Registry from referencing.exceptions import NoSuchResource def unique_object(pairs): result = {} for key, value in pairs: if key in result: raise ValueError("Duplicate JSON object member") result[key] = value return result def reject_constant(_value): raise ValueError("Nonstandard JSON numeric constant") def exact_decimal(value): # Bound work before conversion; preserve non-integral lexical values exactly. # The envelope's only numeric fields are schema integers. A decimal fraction # must never round into an integer and evade schema validation. if len(value) > 128: raise ValueError("JSON numeric token exceeds length bound") try: parsed = Decimal(value) except InvalidOperation as error: raise ValueError("JSON numeric token cannot be represented exactly") from error if not parsed.is_finite() or not (-324 <= parsed.adjusted() <= 308): raise ValueError("JSON numeric exponent outside illustrative bounds") return int(parsed) if parsed == parsed.to_integral_value() else parsed def strict_json(text): return json.loads(text, object_pairs_hook=unique_object, parse_constant=reject_constant, parse_float=exact_decimal) def read_local(directory, name, max_bytes): with (directory / name).open("rb") as source: raw = source.read(max_bytes + 1) if len(raw) > max_bytes: raise ValueError("Local fixture exceeds byte limit: " + name) return strict_json(raw.decode("utf-8")) def deny_remote_schema(uri): raise NoSuchResource(ref=uri) def check_no_references(value): if isinstance(value, dict): if any(key in value for key in ("$ref", "$dynamicRef", "$recursiveRef")): raise ValueError("This offline verifier disallows schema references") for item in value.values(): check_no_references(item) elif isinstance(value, list): for item in value: check_no_references(item) def fingerprint(payload, validator): """Specific to this ASCII-only v2 schema; not a general JSON canonicalizer. Validate first. Normalize the two schema-integer fields to mathematical integers, recursively sort ASCII object keys, serialize compact ASCII JSON without spaces/newline, UTF-8 encode, then SHA-256. Decimal quantity strings are preserved: '2.50' and '2.5' intentionally produce different fingerprints. The authenticated namespace belongs in the durable lookup key separately; caller-supplied stream_id is not authentication. """ validator.validate(payload) normalized = dict(payload) normalized["schema_version"] = int(payload["schema_version"]) normalized["sequence"] = int(payload["sequence"]) encoded = json.dumps(normalized, sort_keys=True, separators=(",", ":"), ensure_ascii=True, allow_nan=False).encode("utf-8") return hashlib.sha256(encoded).hexdigest() def temporal_accept(event_text, expiry_text, now_text, validator): # Reuse date-time schema and FormatChecker before parsing/comparing dates. for text in (event_text, expiry_text, now_text): Draft202012Validator(validator.schema["properties"]["event_time_utc"], format_checker=FormatChecker()).validate(text) event, expiry, now = [datetime.fromisoformat(text.replace("Z", "+00:00")) for text in (event_text, expiry_text, now_text)] return (timedelta(0) < expiry - event <= timedelta(seconds=120) and event - now <= timedelta(seconds=5) and now - event <= timedelta(seconds=120) and now < expiry) def require(condition, case_name): if not condition: raise ValueError("Fixture mismatch: " + case_name) def retry_plan(attempts_completed, elapsed_ms, validity_remaining_ms, outcome, jitter_ms): """Pure record-only worker planner. Does not wait, send or reattempt anything. Caller supplies monotonic elapsed time and a newly checked wall-clock expiry remainder. Every retry requires a confirmed transient no-commit outcome. Unknown outcomes require receipt reconciliation, not blind retransmission. Reserve a 2-second attempt cap inside both the 15-second total budget and exclusive expiry. Caller must recheck deadlines immediately before acting. """ values = (attempts_completed, elapsed_ms, validity_remaining_ms, jitter_ms) if any(type(value) is not int for value in values): raise ValueError("Retry planner requires integer millisecond/count inputs") if not (1 <= attempts_completed <= 4 and 0 <= elapsed_ms <= 86400000 and -86400000 <= validity_remaining_ms <= 86400000 and 0 <= jitter_ms <= 250): raise ValueError("Retry planner input outside illustrative bounds") if outcome not in ("transient_no_commit", "unknown", "success", "permanent_failure"): raise ValueError("Unknown prior-attempt outcome") if outcome == "unknown": return {"action": "reconcile", "reason": "receipt_unknown"} if outcome != "transient_no_commit": return {"action": "stop", "reason": outcome} if attempts_completed == 4: return {"action": "stop", "reason": "attempt_limit"} delay_ms = (1000, 2000, 4000)[attempts_completed - 1] + jitter_ms if elapsed_ms + delay_ms + 2000 > 15000: return {"action": "stop", "reason": "total_deadline"} if delay_ms + 2000 >= validity_remaining_ms: return {"action": "stop", "reason": "exclusive_expiry"} return {"action": "retry", "delay_ms": delay_ms, "attempt_timeout_ms": 2000} def main(): directory = Path(__file__).resolve().parent prefix = "trading-webhook-record-only." schema = read_local(directory, prefix + "schema.json", 131072) example = read_local(directory, prefix + "example.json", 16384) fixtures = read_local(directory, prefix + "fixtures.json", 524288) require(schema.get("$schema") == "https://json-schema.org/draft/2020-12/schema", "supported schema draft") check_no_references(schema) Draft202012Validator.check_schema(schema) validator = Draft202012Validator(schema, format_checker=FormatChecker(), registry=Registry(retrieve=deny_remote_schema)) validator.validate(example) counts = {"schema_cases": 0, "temporal_cases": 0, "parser_cases": 0, "fingerprint_cases": 0, "retry_cases": 0, "lexical_schema_cases": 0} for case in fixtures["schema_cases"]: actual = validator.is_valid(case["payload"]) require(actual == case["expected_schema_valid"], case["name"]) counts["schema_cases"] += 1 for case in fixtures["lexical_schema_cases"]: actual = validator.is_valid(strict_json(case["raw_json"])) require(actual == case["expected_schema_valid"], case["name"]) counts["lexical_schema_cases"] += 1 for case in fixtures["temporal_cases"]: actual = temporal_accept(case["event_time_utc"], case["valid_until_utc"], case["receiver_time_utc"], validator) require(actual == case["expected_temporal_accept"], case["name"]) counts["temporal_cases"] += 1 for case in fixtures["raw_json_rejection_cases"]: try: strict_json(case["raw_json"]) rejected = False except ValueError: rejected = True require(rejected == case["expected_parser_reject"], case["name"]) counts["parser_cases"] += 1 for case in fixtures["fingerprint_cases"]: payload = strict_json(case["raw_json"]) actual = fingerprint(payload, validator) require(actual == case["expected_sha256"], case["name"]) require((actual == fingerprint(example, validator)) == case["same_as_example"], case["name"] + " equivalence") counts["fingerprint_cases"] += 1 for case in fixtures["retry_cases"]: try: actual = retry_plan(**case["inputs"]) except ValueError: actual = {"action": "reject_invalid_input"} require(actual == case["expected"], case["name"]) counts["retry_cases"] += 1 print(json.dumps({"status": "passed", **counts, "example_sha256": fingerprint(example, validator)}, indent=2)) if __name__ == "__main__": try: main() except (OSError, ValueError, KeyError, TypeError, RecursionError, SchemaError, ValidationError) as error: print("Offline verification failed: " + str(error), file=sys.stderr) sys.exit(1)