#!/usr/bin/env python3 """TWMD proof verifier — check that a row you hold was in the snapshot TWMD published. THIS FILE IS FOR CUSTOMERS TO COPY. It imports nothing from `feature_engine` and nothing from PyPI: Python's standard library only, plus an OPTIONAL `cryptography` for the Ed25519 signature check. A verifier that needs our package to run would be asking you to trust our package, which is exactly what verification is supposed to remove. WHAT IT PROVES integrity the row you hold is byte-for-byte the row committed in that snapshot origin the root came from TWMD (only when the signature check passes) WHAT IT DOES NOT PROVE correctness if TWSE published a wrong figure, TWMD faithfully committed to the wrong figure. A valid proof over a wrong number is a valid proof over a wrong number. USAGE # fetch the proof from the public endpoint and verify a row you hold python twmd_verify_proof.py --api https://api.twmarketdata.com \\ --dataset twse_daily_price --row-key '2330|2026-08-14' \\ --row-json '{"ticker":"2330","close":1085.0,"trade_date":"2026-08-14"}' # verify a proof you already saved, offline, with no network at all python twmd_verify_proof.py --proof-file proof.json --row-file row.json \\ --public-key-file twmd.pem # check only that the proof is internally consistent (no row of your own) python twmd_verify_proof.py --proof-file proof.json EXIT CODES 0 verified 1 NOT verified (a mismatch — this is the interesting failure) 2 could not check (network, malformed input, missing key) """ from __future__ import annotations import argparse import base64 import datetime as _dt import hashlib import json import re import sys import urllib.parse import urllib.request LEAF_PREFIX = b"\x00" NODE_PREFIX = b"\x01" #: The algorithm THIS file implements. A proof stamped with anything else is refused with exit 2 #: ("could not check"), never verified with the rules below. #: #: This is the difference between a useful verifier and a harmful one. If TWMD ever changes the hash #: function, the leaf encoding or the path direction, a verifier that plows ahead computes a #: different root and prints "NOT verified" — and the reader concludes the DATA was tampered with. #: A false alarm about integrity is worse than no verifier at all, so a version it does not know is #: an "I cannot check this", not a finding. SUPPORTED_ALGO = "sha256/merkle-v1" SUPPORTED_RECIPE_VERSION = 1 _TICKER_KEYS = {"ticker", "tickers", "symbol", "symbols", "security_code", "security_codes", "code", "codes"} # --------------------------------------------------------------------------------------------- # canonicalisation — MUST match feature_engine/read_api/query_identity.canonicalize_params # --------------------------------------------------------------------------------------------- def _normalise_scalar(value): if isinstance(value, bool): # before int: bool is an int subclass return value if isinstance(value, _dt.datetime): moment = value if value.tzinfo else value.replace(tzinfo=_dt.timezone.utc) return moment.astimezone(_dt.timezone.utc).isoformat().replace("+00:00", "Z") if isinstance(value, _dt.date): return value.isoformat() if isinstance(value, (int, float, str)): return value return str(value) def _normalise_value(key, value): if isinstance(value, dict): return {k: _normalise_value(k, v) for k, v in sorted(value.items(), key=lambda kv: str(kv[0]))} if isinstance(value, (list, tuple, set)): items = [_normalise_value(key, v) for v in value] return sorted(items, key=lambda v: json.dumps(v, sort_keys=True, ensure_ascii=False)) if str(key).lower() in _TICKER_KEYS and isinstance(value, str): return value.strip().upper() return _normalise_scalar(value) def canonicalize_row(row): """Drop empty/None entries, keep False/0, sort keys, fold tickers, ISO-format dates. Every rule here exists because the same row arrives spelled differently. `False` and `0` are KEPT — they are values; dropping them would make `no_trade=false` hash the same as an absent `no_trade`, which is a different row. """ canonical = {} for key in sorted(row or {}, key=str): value = row[key] if value is None: continue if isinstance(value, str) and not value.strip(): continue if isinstance(value, (list, tuple, set)) and len(value) == 0: continue canonical[str(key)] = _normalise_value(str(key), value) return canonical def canonical_row_bytes(row): return json.dumps(canonicalize_row(row), sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8") # --------------------------------------------------------------------------------------------- # the verifier — this is the whole thing # --------------------------------------------------------------------------------------------- def leaf_hash(row): return hashlib.sha256(LEAF_PREFIX + canonical_row_bytes(row)).hexdigest() def node_hash(left, right): return hashlib.sha256(NODE_PREFIX + bytes.fromhex(left) + bytes.fromhex(right)).hexdigest() def verify_inclusion(leaf, path, root): """Recompute the root from a leaf and its sibling path. Ten lines; that is the point. `side` is the side of the SIBLING, and it matters because node_hash is order-sensitive. """ computed = str(leaf) for step in path: sibling = str(step["hash"]) side = str(step.get("side", "right")) computed = node_hash(sibling, computed) if side == "left" else node_hash(computed, sibling) return computed == str(root) def verify_signature(root, signature_b64, public_key_pem): """Ed25519 over the ASCII hex of the root. Returns (checked, ok, detail). `checked` is False when the check could not be RUN (no key, `cryptography` not installed) — a distinct outcome from a signature that was checked and failed, because one is your setup and the other is a red alert. """ if not signature_b64: return False, False, "checkpoint is unsigned (integrity still provable; origin is not)" if not public_key_pem: return False, False, "no public key supplied; pass --public-key-file or --api" try: from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives import serialization except ImportError: return False, False, ("`cryptography` is not installed, so the signature was not checked. " "The Merkle path above is still fully verified.") try: key = serialization.load_pem_public_key(public_key_pem.encode("ascii")) key.verify(base64.b64decode(signature_b64), str(root).encode("ascii")) return True, True, "signature valid" except InvalidSignature: return True, False, "SIGNATURE INVALID — this root did not come from that key" except (ValueError, TypeError) as exc: return False, False, f"could not check the signature: {exc}" # --------------------------------------------------------------------------------------------- # fetching # --------------------------------------------------------------------------------------------- def _get_json(url, timeout=20): request = urllib.request.Request(url, headers={"Accept": "application/json", "User-Agent": "twmd-verify-proof/1"}) with urllib.request.urlopen(request, timeout=timeout) as response: # noqa: S310 - http(s) only return json.loads(response.read().decode("utf-8")) def _require_http_url(url, what): parts = urllib.parse.urlsplit(str(url)) if parts.scheme not in {"http", "https"}: raise SystemExit(f"{what} must be an http(s) URL, got {url!r}") return str(url).rstrip("/") def fetch_proof(api, dataset, row_key, snapshot_version=None): base = _require_http_url(api, "--api") query = {"dataset": dataset, "row_key": row_key} if snapshot_version: query["snapshot_version"] = snapshot_version return _get_json(f"{base}/v2/proof/inclusion?{urllib.parse.urlencode(query)}") def fetch_public_key(api): base = _require_http_url(api, "--api") return (_get_json(f"{base}/v2/proof/public-key") or {}).get("public_key_pem") # --------------------------------------------------------------------------------------------- def _load_json_arg(inline, path, what): if inline and path: raise SystemExit(f"pass either --{what}-json or --{what}-file, not both") if path: with open(path, "r", encoding="utf-8") as handle: return json.load(handle) if inline: return json.loads(inline) return None def main(argv=None): parser = argparse.ArgumentParser( description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) parser.add_argument("--api", help="TWMD API base URL (fetches the proof and the public key)") parser.add_argument("--dataset") parser.add_argument("--row-key", help="the snapshot's logical key, e.g. '2330|2026-08-14'") parser.add_argument("--snapshot-version", default=None, help="omit for the most recent checkpoint") parser.add_argument("--proof-file", help="a proof saved earlier (fully offline verification)") parser.add_argument("--row-json", help="the row you hold, as JSON") parser.add_argument("--row-file", help="the row you hold, as a JSON file") parser.add_argument("--public-key-file", help="TWMD's Ed25519 public key, PEM") parser.add_argument("--quiet", action="store_true") args = parser.parse_args(argv) # -- obtain the proof if args.proof_file: with open(args.proof_file, "r", encoding="utf-8") as handle: proof = json.load(handle) elif args.api and args.dataset and args.row_key: try: proof = fetch_proof(args.api, args.dataset, args.row_key, args.snapshot_version) except Exception as exc: # noqa: BLE001 - any fetch problem is "could not check" print(f"could not fetch the proof: {exc}", file=sys.stderr) return 2 else: parser.error("supply --proof-file, or --api with --dataset and --row-key") return 2 status = str(proof.get("status", "")) if status != "ok": # `not_in_snapshot` is a TRUE answer, not a failure of the tool — say which it was. print(json.dumps({"verified": False, "status": status, "message": proof.get("message", "no proof was returned")}, indent=2, ensure_ascii=False)) return 1 if status == "not_in_snapshot" else 2 # -- refuse an algorithm this file does not implement, BEFORE computing anything algo = str(proof.get("algo") or SUPPORTED_ALGO) if algo != SUPPORTED_ALGO: print(json.dumps({ "verified": None, "status": "unsupported_algorithm", "proof_algo": algo, "this_verifier_implements": SUPPORTED_ALGO, "message": "this proof uses an algorithm this verifier does not implement. It has NOT " "been checked. Do not read this as a verification failure — update the " "verifier (see /v2/proof/recipe) and check again.", }, indent=2, ensure_ascii=False)) return 2 root = proof["root"] claimed_leaf = proof["leaf_hash"] path = proof.get("merkle_path") or [] report = { "dataset": proof.get("dataset"), "row_key": proof.get("row_key"), "snapshot_version": (proof.get("checkpoint") or {}).get("snapshot_version"), "root": root, "leaf_count": (proof.get("checkpoint") or {}).get("leaf_count"), "path_length": len(path), } # -- 1. does the published path actually reach the published root? path_ok = verify_inclusion(claimed_leaf, path, root) report["merkle_path_reaches_root"] = path_ok # -- 2. does the row YOU hold hash to that leaf? row = _load_json_arg(args.row_json, args.row_file, "row") if row is not None: own_leaf = leaf_hash(row) report["your_row_leaf_hash"] = own_leaf report["your_row_matches_the_committed_leaf"] = (own_leaf == claimed_leaf) if own_leaf != claimed_leaf: report["note"] = ("your row hashes to a different leaf. Either it is not the row that " "was committed, or a field differs (check types: 1085 vs 1085.0, and " "date spellings). Canonicalisation rules are in --help.") else: report["your_row_matches_the_committed_leaf"] = None report["note"] = ("no row supplied, so only the proof's internal consistency was checked. " "Pass --row-json/--row-file to verify a row you hold.") # -- 3. did the root come from TWMD? public_key = None if args.public_key_file: with open(args.public_key_file, "r", encoding="utf-8") as handle: public_key = handle.read() elif args.api: try: public_key = fetch_public_key(args.api) except Exception as exc: # noqa: BLE001 report["signature_detail"] = f"could not fetch the public key: {exc}" checked, sig_ok, detail = verify_signature(root, proof.get("signature"), public_key) report["signature_checked"] = checked report["signature_valid"] = sig_ok if checked else None report["signature_detail"] = detail verified = bool(path_ok) and (report["your_row_matches_the_committed_leaf"] is not False) \ and (not checked or sig_ok) report["verified"] = verified report["proves"] = "integrity and origin, NOT semantic correctness" if not args.quiet: print(json.dumps(report, indent=2, ensure_ascii=False)) return 0 if verified else 1 if __name__ == "__main__": raise SystemExit(main())