#!/usr/bin/env python3 # SPDX-License-Identifier: MIT — see /clients/LICENSE. Copy me freely. """kaspulse.py — a tiny zero-dependency VERIFYING client for the kaspulse oracle. Python 3.9+, stdlib only (hashlib + urllib). No pip installs, no keys. from kaspulse import Kaspulse k = Kaspulse('https://pulse.kascov.io') feed = k.feed('KAS/USD') pin = load_previously_authenticated_committee() # not from this feed origin r = k.verify_with_committee(feed, pin) if not r['ok']: raise ValueError(r.get('error', 'untrusted feed')) c = k.verify_covenant(feed) # the BOUND kaspulse/cov/v2 covenant preimage px = k.checked_value(feed) # verified + fresh + not halted, or ValueError L1 KCC20 (Kaspa L1 AMM pools, priced through kascov) is first-class: rows = k.kcc20_feeds() # the kind == "kcc20-pool" catalog rows info = k.kcc20_info(feed) # basis, last_trade_age_s, anchor_txid, # unpriced_reason, gates, depth — flattened r = k.re_derive_kcc20(feed) # re-fetch kascov and demand the exact # integer rational back (network call) cen = k.kcc20_census() # every market kaspulse REFUSED, with the reason Honest scope, in four parts. (1) verify_feed checks the committee's signatures (3-of-5 BIP340 Schnorr over blake2b-256 of "kaspulse/v2|PAIR|mant|expo|ts|round"), the binding of the signed message to the JSON fields, and the safety flags ('halted', 'peg_ok') — offline. It does NOT check FRESHNESS: nothing there compares 'signed_ts' to a clock. That is checked_value's job, and this sentence used to claim otherwise. (2) It does NOT re-fetch the exchanges and recompute the majors' median; compare exchange quotes separately if needed. (3) re_derive_kcc20 DOES re-derive an L1 price from source — but from kascov, so it is a cross-check, not independence; and it does not replay the admission gates. Both limits are spelled out at the KCC20 block. (4) verify_covenant REBUILDS the bound `kaspulse/cov/v2` covenant preimage from the feed's own fields and demands the published bytes back exactly — the check that would have caught the unbound domain withdrawn on 2026-07-27. verify_feed folds it in, so a feed whose covenant does not rebuild NEVER returns ok=True. Both cores self-test at import — the crypto core against BIP340 official vector 0, a corrupted copy that must FAIL, and blake2b-256(b"abc"); the KCC20 core against pinned mainnet tier fixtures. A broken crypto core raises RuntimeError at import (nothing here is usable without it); a broken KCC20 core warns and makes re_derive_kcc20 refuse, because signature verification is a separate concern and stays sound. Neither ever returns a fake verdict. CLI: python3 kaspulse.py verify KAS/USD [base] python3 kaspulse.py kcc20 [base] # re-derive every L1 KCC20 price Publishing to PyPI is a separate decision — this file is the whole client. """ import hashlib import json import math import re import sys import time import urllib.error import urllib.request # ── hashes ────────────────────────────────────────────────────────────────── def blake2b256(data): """Unkeyed blake2b, 32-byte digest — what the oracle hashes the message with.""" return hashlib.blake2b(data, digest_size=32).digest() def tagged_hash(tag, msg): """BIP340 tagged hash: sha256(sha256(tag) || sha256(tag) || msg).""" t = hashlib.sha256(tag.encode('ascii')).digest() return hashlib.sha256(t + t + msg).digest() # ── BIP340 Schnorr verification over secp256k1 (verify-only, public data — no # side-channel concern; follows the BIP340 reference implementation) ─────── _P = 0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFEFFFFFC2F _N = 0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFEBAAEDCE6AF48A03BBFD25E8CD0364141 _G = ( 0x79BE667EF9DCBBAC55A06295CE870B07029BFCDB2DCE28D959F2815B16F81798, 0x483ADA7726A3C4655DA4FBFC0E1108A8FD17B448A68554199C47D08FFB10D4B8, ) def _lift_x(x): """The curve point with this x and EVEN y, or None (BIP340 lift_x).""" if x >= _P: return None c = (pow(x, 3, _P) + 7) % _P y = pow(c, (_P + 1) // 4, _P) if pow(y, 2, _P) != c: return None return (x, y if y % 2 == 0 else _P - y) def _point_add(p1, p2): """Affine addition; None = point at infinity.""" if p1 is None: return p2 if p2 is None: return p1 x1, y1 = p1 x2, y2 = p2 if x1 == x2 and (y1 + y2) % _P == 0: return None # P + (−P) if p1 == p2: lam = (3 * x1 * x1 * pow(2 * y1, -1, _P)) % _P else: lam = ((y2 - y1) * pow(x2 - x1, -1, _P)) % _P x3 = (lam * lam - x1 - x2) % _P return (x3, (lam * (x1 - x3) - y1) % _P) def _point_mul(pt, k): """Double-and-add scalar multiplication.""" r = None while k > 0: if k & 1: r = _point_add(r, pt) pt = _point_add(pt, pt) k >>= 1 return r def bip340_verify(pubkey32, msg32, sig64): """Standard BIP340 verification. pubkey32 = x-only key, msg32 = the 32-byte message (for kaspulse: the blake2b-256 digest — NOT hashed again outside BIP340's own tagged hashing), sig64 = r||s. Returns bool, never raises.""" if len(pubkey32) != 32 or len(msg32) != 32 or len(sig64) != 64: return False pt = _lift_x(int.from_bytes(pubkey32, 'big')) if pt is None: return False r = int.from_bytes(sig64[:32], 'big') s = int.from_bytes(sig64[32:], 'big') if r >= _P or s >= _N: return False e = int.from_bytes(tagged_hash('BIP0340/challenge', sig64[:32] + pubkey32 + msg32), 'big') % _N R = _point_add(_point_mul(_G, s), _point_mul(pt, _N - e)) # s·G − e·P if R is None or R[1] % 2 != 0 or R[0] != r: return False return True # ── the kaspulse/v2 signed message and the feed verdict ───────────────────── def parse_signed_message(message): """message = "kaspulse/v2|PAIR|mant|expo|ts|round" (ASCII, decimal ints, expo may be negative). Returns the five fields AS STRINGS (no float round-trips) or None if the shape is wrong.""" if not isinstance(message, str): return None parts = message.split('|') if len(parts) != 6 or parts[0] != 'kaspulse/v2': return None pair, mant, expo, ts, rnd = parts[1:] if (not pair or not re.fullmatch(r'\d+', mant) or not re.fullmatch(r'-?\d+', expo) or not re.fullmatch(r'\d+', ts) or not re.fullmatch(r'\d+', rnd)): return None return {'pair': pair, 'mant': mant, 'expo': expo, 'ts': ts, 'round': rnd} def _hex_bytes(s): """Lowercase-hex canonical, tolerant of uppercase; None on junk.""" if not isinstance(s, str) or len(s) % 2 != 0: return None try: return bytes.fromhex(s) except ValueError: return None def _message_binds(parsed, feed): """The FIVE-field binding, factored out so the import-time self-test can drive it directly. `round` is on the list because it used to be parsed and DROPPED: a lying or compromised API could serve any signed_round it liked and this verifier still returned bound=True, while the server-side verifier rejected the identical feed. The covenant rebuild does not cover that gap for a feed publishing no covenant — which is every archived feed.""" # string-compare the integers — no float round-trips return (parsed is not None and parsed['pair'] == feed.get('pair') and parsed['mant'] == str(feed.get('mant')) and parsed['expo'] == str(feed.get('expo')) and parsed['ts'] == str(feed.get('signed_ts')) and parsed['round'] == str(feed.get('signed_round'))) def _count_distinct_valid(signers, sigs, digest, results): """Count the signatures that verify over `digest`, crediting each DISTINCT signer AT MOST ONCE, and append one {'signer','ok','duplicate'} row per list entry. The dedupe is the point. Counting list ENTRIES let one genuine key, listed three times with its one genuine signature, satisfy 3-of-5 — so a single compromised committee key plus control of the JSON (the API host, a CDN, an untrusted mirror) forged any price and every kaspulse verifier called it threshold-signed. The on-chain m-of-n gate was never affected (the redeem bakes distinct keys into fixed slots); this was an off-chain-only threshold collapse, which is exactly where "3-of-5, verifiable by anyone" lives. Deduped on the DECODED 32 bytes, so a re-cased hex string is the same key.""" seen = set() valid = 0 for i, signer in enumerate(signers): pk = _hex_bytes(signer) sig = _hex_bytes(sigs[i]) if i < len(sigs) else None verifies = (pk is not None and sig is not None and len(pk) == 32 and len(sig) == 64 and bip340_verify(pk, digest, sig)) duplicate = pk is not None and pk in seen ok = verifies and not duplicate if pk is not None: seen.add(pk) results.append({'signer': str(signer), 'ok': ok, 'duplicate': duplicate}) if ok: valid += 1 return valid def verify_message(feed): """The MESSAGE half of the verdict, on its own. Pure, no network. VALID := (count of BIP340-verifying signatures ≥ threshold) AND the signed message's PAIR/mant/expo/ts/round equal the JSON's pair/mant/expo/signed_ts/signed_round. Returns {'ok', 'valid', 'threshold', 'bound', 'parsed', 'results'} (+'error'). `verify_feed` below wraps this and also folds in the covenant domain — use THAT unless you specifically want the message half alone.""" if not isinstance(feed, dict) or not isinstance(feed.get('message'), str): return {'ok': False, 'valid': 0, 'threshold': 0, 'bound': False, 'parsed': None, 'results': [], 'error': 'not a feed object (need message/signers/signatures)'} signers = feed.get('signers') if isinstance(feed.get('signers'), list) else [] sigs = feed.get('signatures') if isinstance(feed.get('signatures'), list) else [] threshold = feed.get('threshold') if isinstance(feed.get('threshold'), int) and feed.get('threshold') > 0 else 0 parsed = parse_signed_message(feed['message']) # field binding: what was SIGNED must equal what the JSON claims # (string-compare the integers — no float round-trips) bound = _message_binds(parsed, feed) digest = blake2b256(feed['message'].encode('ascii', errors='replace')) results = [] valid = _count_distinct_valid(signers, sigs, digest, results) ok = bound and threshold > 0 and valid >= threshold out = {'ok': ok, 'valid': valid, 'threshold': threshold, 'bound': bound, 'parsed': parsed, 'results': results} if parsed is None: out['error'] = 'unparsable signed message (want kaspulse/v2|PAIR|mant|expo|ts|round)' elif not bound: out['error'] = 'signed message fields do not match the JSON fields (pair/mant/expo/signed_ts/signed_round)' elif threshold == 0: out['error'] = 'missing threshold' elif valid < threshold: out['error'] = 'only %d of %d required signatures verify' % (valid, threshold) return out # ── the BOUND covenant domain: kaspulse/cov/v2 ────────────────────────────── # # WHY THIS EXISTS. Until 2026-07-27 the oracle also signed a COVENANT preimage: # blake2b(price_bytes(price_e8)) — a bare integer with no pair, no exponent, no # round and no timestamp. Two live consequences, both real: # # * BTC/USD's five committee signatures unlocked ANY on-chain price gate with # a lower strike, on ANY pair. The signatures were genuine; they just did # not say WHAT they were about. # * three feeds quantized to price_e8 = 0 and published real signatures over # blake2b(empty), which satisfies every "at or below" gate forever. # # The field was withdrawn and is back only under `kaspulse/cov/v2`: ONE blob, # # "kaspulse/cov/v2" 15 bytes ASCII 0..15 ┐ # blake2b256(PAIR)[0..8] 8 bytes 15..23 ├ bound prefix, baked # expo 1 byte (i8) 23..24 ┘ into the P2SH script # round 8 bytes (u64 BE) 24..32 # ts 8 bytes (u64 BE) 32..40 # mant minimal-LE script number, variable, 40..end # # signed as schnorr(blake2b256(blob)). # # THE CHECK BELOW IS THE ONE THAT WOULD HAVE CAUGHT THE ORIGINAL HOLE. It never # parses the oracle's `covenant.preimage` and trusts what it finds — it REBUILDS # the blob from the feed's own pair/mant/expo/signed_round/signed_ts and demands # the published bytes back EXACTLY. A cross-pair blob, a shifted exponent, a # replayed round, a truncated tail: all of them are a different byte string, and # a different byte string is a hard FAIL before a single signature is checked. # Only then are the signatures verified — against the digest of OUR blob. # # Byte-identical logic to clients/js/kaspulse.mjs and web/vendor/verify.js. COV_V2_TAG = b'kaspulse/cov/v2' # exactly 15 ASCII bytes COV_V2_PREFIX_LEN = 24 # tag ‖ pair-hash ‖ expo COV_V2_MANT_OFF = 40 # ‖ round_be ‖ ts_be, then the mantissa def cov_v2_pair_id(pair): """blake2b256(PAIR)[0..8] — the pair binding. The RAW pair string is hashed ("KAS/USD"), not a domain-prefixed one: the 15-byte tag in front already separates this domain from the bond record's.""" return blake2b256(pair.encode('utf-8'))[:8] def cov_v2_mant_bytes(mant): """Minimal little-endian script-number encoding of a POSITIVE mantissa — the variable-length tail, and the operand the on-chain script compares to the strike. A trailing byte with the top bit set would read as NEGATIVE on the VM, so it gets a zero sign byte.""" out = bytearray() v = mant while v > 0: out.append(v & 0xFF) v >>= 8 if out and out[-1] & 0x80: out.append(0) return bytes(out) def _cov_v2_int(v, signed): """Strict decimal -> int. String-parsed on purpose: a JSON value that arrived as a float or as 2**53+1 must not silently become an integer.""" s = str(v) if not re.fullmatch(r'-?\d+' if signed else r'\d+', s): return None return int(s) def cov_v2_preimage(pair, mant, expo, round_, ts): """Build the bound preimage from raw fields. Returns bytes, or None when the fields cannot form a LEGAL cov/v2 blob — which is itself a verdict: * expo must fit one signed byte (live feeds sit in -12..3); * mant must be NON-ZERO (a zero mantissa is an empty tail, which the VM reads as numeric zero — the blake2b(empty) hole, one layer down); * mant must fit a signed 8-byte script number, or the on-chain compare would abort; * round and ts must fit u64.""" if not isinstance(pair, str) or not pair: return None m = _cov_v2_int(mant, False) e = _cov_v2_int(expo, True) r = _cov_v2_int(round_, False) t = _cov_v2_int(ts, False) if m is None or e is None or r is None or t is None: return None if not -128 <= e <= 127: return None if m == 0 or m > 0x7FFFFFFFFFFFFFFF: return None if r > 0xFFFFFFFFFFFFFFFF or t > 0xFFFFFFFFFFFFFFFF: return None prefix = bytearray(COV_V2_PREFIX_LEN) prefix[:15] = COV_V2_TAG prefix[15:23] = cov_v2_pair_id(pair) prefix[23] = e & 0xFF # i8 two's complement return bytes(prefix) + r.to_bytes(8, 'big') + t.to_bytes(8, 'big') + cov_v2_mant_bytes(m) def verify_covenant(feed): """Verify a feed's cov/v2 covenant attestation. Pure, no network. Returns {'present', 'ok', 'valid', 'threshold', 'matches', 'expected', 'published', 'results'} (+'error'). 'present': False means the feed simply carries no covenant attestation — that is NOT a failure, and a UI should say "no covenant" rather than ✗. Every other False is a real refusal. 'matches' is the byte-for-byte verdict on the preimage and is reported separately from the signature count on purpose: a feed can carry five perfectly valid signatures over the WRONG blob, and that is exactly the shape the withdrawn domain had.""" out = {'present': False, 'ok': False, 'valid': 0, 'threshold': 0, 'matches': False, 'expected': None, 'published': None, 'results': []} if not isinstance(feed, dict): out['error'] = 'not a feed object' return out cov = feed.get('covenant') if isinstance(feed.get('covenant'), dict) else None published = cov.get('preimage', '').lower() if cov and isinstance(cov.get('preimage'), str) else '' sigs = cov.get('signatures') if cov and isinstance(cov.get('signatures'), list) else [] if cov is None or (published == '' and not sigs): out['error'] = 'feed carries no kaspulse/cov/v2 covenant attestation' return out out['present'] = True out['published'] = published or None # (a) the v2 MESSAGE must verify first. A covenant ✓ on a feed whose own # signatures do not verify would be a green tick over nothing. base = verify_message(feed) if not base['ok']: out['error'] = 'feed signatures/binding failed first: ' + str(base.get('error', 'invalid')) return out # (b) signatures with no preimage = a pre-withdrawal artifact from the # unbound blake2b(price_bytes) domain. Never verify those. if published == '': out['error'] = ('covenant.signatures with NO covenant.preimage — pre-withdrawal artifact ' 'of the unbound blake2b(price_bytes) domain; refusing') return out # (c) REBUILD, then compare bytes. Nothing here reads the published blob's # own fields — that would be asking the document to grade itself. pre = cov_v2_preimage(feed.get('pair'), feed.get('mant'), feed.get('expo'), feed.get('signed_round'), feed.get('signed_ts')) if pre is None: out['error'] = ("this feed's own fields cannot form a legal cov/v2 preimage " "(expo outside i8, zero or oversized mant, or non-integer round/ts)") return out out['expected'] = pre.hex() out['matches'] = out['expected'] == published if not out['matches']: # the two blobs are in 'expected' / 'published' — the caller renders them; an # error string carrying 88 hex chars twice is unreadable in every UI there is out['error'] = ("COVENANT PREIMAGE MISMATCH — the published blob is not this feed's " '(cross-pair, wrong-expo or replayed-round signature reuse). ' 'Do NOT fund a covenant against it.') return out # (d) threshold of committee signatures over blake2b256(OUR blob). threshold = feed.get('threshold') if isinstance(feed.get('threshold'), int) and feed.get('threshold') > 0 else 0 out['threshold'] = threshold digest = blake2b256(pre) signers = feed.get('signers') if isinstance(feed.get('signers'), list) else [] valid = _count_distinct_valid(signers, sigs, digest, out['results']) out['valid'] = valid if threshold == 0: out['error'] = 'missing threshold' return out if valid < threshold: out['error'] = ('covenant: only %d of %d required signatures verify over blake2b256(preimage)' % (valid, threshold)) return out out['ok'] = True return out def verify_feed(feed): """THE VERDICT the CLI, the client and the browser button all read. ok := the v2 message verifies AND field-binds, AND the feed's own safety flags are clear ('halted', 'peg_ok'), AND (the feed carries no covenant attestation OR that attestation rebuilds byte-for-byte and carries a threshold of signatures). THE SAFETY FLAGS ARE PART OF THE VERDICT. They were not until 2026-08-25, and the consequence was that a HALTED feed — the circuit breaker doing its job, holding a stale price — came back ok=True with a green ✓ and a price here and in the JS client, while the server-side verifiers refused it. Identical bytes produced inconsistent verdicts; all public verification surfaces now enforce the safety flags. 'halted' is a REFUSAL, not an accusation: the signatures are fine and the committee is honest, the price is simply not safe to consume. So 'halted' and 'depegged' are set on the result as their own keys, and every renderer must show that third state separately from an invalid signature. The covenant clause is strictly ADDITIVE — a feed with no covenant is unaffected, which is every feed published between the 2026-07-27 withdrawal and the cov/v2 ship. It can only turn a ✓ into a ✗, never the reverse. That direction is deliberate: a feed that publishes a covenant preimage which is not its own is forged or compromised, and NOTHING about it should show a green tick. The message-only verdict is verify_message(); the covenant half is under the returned 'covenant' key and in verify_covenant().""" r = verify_message(feed) c = verify_covenant(feed) r['covenant'] = c if c['present'] and not c['ok']: r['ok'] = False r['error'] = 'covenant domain (kaspulse/cov/v2) FAILED: ' + str(c.get('error', 'invalid')) # the safety flags, last, so their message is the one a caller sees: the # signatures may be perfect and the price still not safe to consume. r['halted'] = isinstance(feed, dict) and bool(feed.get('halted')) r['depegged'] = isinstance(feed, dict) and feed.get('peg_ok') is False if r['halted'] or r['depegged']: r['ok'] = False r['error'] = ('feed halted (circuit breaker)' if r['halted'] else 'chain depegged (peg_ok=false)') return r _FUTURE_TOLERANCE_S = 30.0 def _max_age_error(max_age_s): if isinstance(max_age_s, bool) or not isinstance(max_age_s, (int, float)): return 'max_age_s must be a finite, non-negative number' if not math.isfinite(float(max_age_s)) or max_age_s < 0: return 'max_age_s must be a finite, non-negative number' return None def _freshness_error(signed_ts, max_age_s, now_s=None): max_error = _max_age_error(max_age_s) if max_error: return max_error try: signed = float(signed_ts) except (TypeError, ValueError, OverflowError): return 'signed_ts must be a finite timestamp' if not math.isfinite(signed): return 'signed_ts must be a finite timestamp' now = time.time() if now_s is None else now_s age = now - signed if age < -_FUTURE_TOLERANCE_S: return ('signature timestamp is %.1f s in the future (max %.0f s)' % (-age, _FUTURE_TOLERANCE_S)) if age > max_age_s: return 'signature is %.1f s old (max %s s)' % (age, max_age_s) return None def checked_value(feed, max_age_s=30): """The verified price (mant × 10^expo) — or ValueError with the reason it is unsafe: failed verification, halted, depegged, older than max_age_s, or signed more than 30 seconds in the future.""" max_error = _max_age_error(max_age_s) if max_error: raise ValueError('kaspulse: ' + max_error) r = verify_feed(feed) if not r['ok']: raise ValueError('kaspulse: refusing value: ' + r.get('error', 'verification failed')) if feed.get('halted'): raise ValueError('kaspulse: refusing value: feed halted (circuit breaker)') if feed.get('peg_ok') is False: raise ValueError('kaspulse: refusing value: chain depegged (peg_ok=false)') freshness = _freshness_error(feed['signed_ts'], max_age_s) if freshness: raise ValueError('kaspulse: refusing value: ' + freshness) return int(feed['mant']) * 10.0 ** int(feed['expo']) # ── L1 KCC20 re-derivation (Kaspa L1 AMM pools, read from kascov) ─────────── # # WHAT AN L1 PRICE IS. A `kcc20-pool` feed carries an EXACT INTEGER RATIONAL: # `price_num_sompi / price_den`, in sompi per token BASE UNIT. There is no # verified `decimals` for a KCC20 token — only unverified `claimed_decimals` — # so nothing here rescales anything, and neither should you. Python's ints are # arbitrary-precision and json parses them exactly, so the comparison below is # a true integer comparison with no float anywhere near it. # # WHICH TIER PRODUCED IT (the `basis` field), and why it matters: # 'spot' — market.spot_num_sompi / market.spot_den from # /data/{net}/pools: kascov's gated marginal. # 'last_verified_state' — kascov WITHHELD spot, because the covenant's live # token balance does not equal the newest verified # trade's after-balance (an anti-donation gate). It # omits both keys with NO error field, so a consumer # that assumes spot is always present silently drops # feeds, and one that substitutes # reserve / held_by_covenant publishes a number # measured 22.6% wrong on one live pool. The honest # fallback is the newest NON-carried 1h pool candle # close — a verified pool state with a timestamp. # # WHAT re_derive_kcc20 CHECKS: it re-fetches kascov, decides the tier ITSELF # (never from the feed's own `basis` claim, which would be circular), and # demands the two integers back EXACTLY — not within a band. # WHAT IT DOES NOT CHECK: the admission gates (bracket_holds / invariant_holds) # are not replayed here. The oracle performs those checks; this public client # only cross-checks the published rational against kascov. Nothing here is # independent of kascov: a bug inside kascov reproduces here identically. KCC20_NET = 'mainnet' # The kascov origin a re-derivation runs against — PINNED HERE, never taken from # the feed. `price_source_url` is published BY the thing being audited: if it # also chose the server that answers, a forged price served beside a forged # kascov reproduces perfectly and this returns ok=True. So the origin is a # constant, overridable only by the CALLER (`kascov_base`), and the feed's own # URL is CHECKED against it rather than obeyed. The public JavaScript client # applies the same KCC20_PINNED_ORIGIN rule. KCC20_PINNED_ORIGIN = 'https://kascov.io' def _kcc20_int(v): """A JSON integer that may arrive as a number or a decimal string.""" if isinstance(v, bool): return None if isinstance(v, int): return v if isinstance(v, str) and re.fullmatch(r'-?\d+', v): return int(v) return None def _kcc20_origin(url): """The scheme+host of a URL, without a urlparse round-trip.""" if not isinstance(url, str) or '://' not in url: return None i = url.index('://') + 3 j = url.find('/', i) return url if j < 0 else url[:j] def kcc20_info(feed): """Every L1 provenance field on a FeedObj, flattened. None for any feed that is not kind == "kcc20-pool". `claimed_*` are DEPLOYER-SUPPLIED AND UNVERIFIED and must be labelled as such wherever they are shown.""" if not isinstance(feed, dict) or feed.get('kind') != 'kcc20-pool': return None g = feed.get return { 'pair': g('pair'), 'leg': g('leg'), 'token_covenant_id': g('token_covenant_id'), 'market_covenant_id': g('market_covenant_id'), 'display_name': g('display_name'), 'listed_name': g('listed_name'), 'listed_ticker': g('listed_ticker'), 'listed_decimals': g('listed_decimals'), 'listed_known': g('listed_known') is True, 'listed_checks_passed': g('listed_checks_passed') is True, 'listed_list_name': g('listed_list_name'), 'listed_art_url': g('listed_art_url'), 'claimed_ticker': g('claimed_ticker'), 'claimed_name': g('claimed_name'), 'claimed_decimals': g('claimed_decimals'), 'unverified_note': 'claimed_* are deployer-supplied and unverified; a listed_* row ' 'is BOUND to its covenant (creator key, genesis tx and curve all ' "match the chain) but its name and ticker are the registry's " 'labels, not on-chain facts; every signed L1 price remains per ' 'token BASE UNIT', 'basis': g('basis'), 'basis_note': g('basis_note'), 'num_sompi': g('price_num_sompi'), 'den': g('price_den'), 'verified_as_of_ms': g('verified_as_of_ms'), 'last_trade_age_s': g('last_trade_age_s'), 'anchor_txid': g('anchor_txid'), 'anchor_daa': g('anchor_daa'), 'admitted': g('admitted'), 'rejected_bracket': g('rejected_bracket'), 'gate_window': g('gate_window'), 'gate_note': g('gate_note'), 'skeleton': g('skeleton'), 'invariant_ok': g('invariant_ok'), 'reserve_kas': g('reserve_kas'), 'taker_fee_bps': g('taker_fee_bps'), 'thin': g('thin'), 'move_10pct_usd': g('move_10pct_usd'), 'depth_2pct_usd': g('depth_2pct_usd'), 'resting_asks': g('resting_asks'), # non-null means the ORACLE ITSELF could not provenance this price — a # price you must not consume, not a cosmetic field 'unpriced_reason': g('unpriced_reason'), 'price_source_url': g('price_source_url'), } def kcc20_tier_of(pools_doc, market_covenant_id): """Which tier applies, decided from kascov's /pools document alone. Pure. Returns {'found', 'basis', 'num', 'den', 'row'}.""" rows = pools_doc.get('pools') if isinstance(pools_doc, dict) else None row = None for p in (rows or []): if isinstance(p, dict) and p.get('market_id') == market_covenant_id: row = p break if row is None: return {'found': False, 'basis': None, 'num': None, 'den': None, 'row': None} m = row.get('market') or {} n, d = _kcc20_int(m.get('spot_num_sompi')), _kcc20_int(m.get('spot_den')) if n is not None and d is not None and n > 0 and d > 0: return {'found': True, 'basis': 'spot', 'num': n, 'den': d, 'row': row} # spot withheld — silently, by design. Tier 2 needs the candles document. return {'found': True, 'basis': 'last_verified_state', 'num': None, 'den': None, 'row': row} def kcc20_candle_close(candles_doc): """Newest NON-carried candle close as {'num','den','t'}. `carried` rows are gap-fill: they repeat the last close forward at zero volume and would make a dead market look freshly verified. Pure.""" cs = candles_doc.get('candles') if isinstance(candles_doc, dict) else None for c in reversed(cs or []): if not isinstance(c, dict) or c.get('carried'): continue close = c.get('close') or {} num, den = _kcc20_int(close.get('quote_sompi')), _kcc20_int(close.get('base_amount')) if num is not None and den is not None and num > 0 and den > 0: return {'num': num, 'den': den, 't': c.get('t')} return None # the newest non-carried candle is unusable — don't walk past it return None def _kcc20_get(url): req = urllib.request.Request(url, headers={'Accept': 'application/json'}) with urllib.request.urlopen(req, timeout=15) as resp: return json.load(resp) def _kcc20_err(url, e): """A fetch failure must not read as 'the price is wrong'. In particular a stdlib TLS failure is the single most common false alarm for this client — python.org macOS builds ship with no root store until Install Certificates.command is run — so it is named, not left as a wall of OpenSSL. Never a suggestion to turn verification off.""" msg = 'kascov: %s: %s' % (url, e) if 'CERTIFICATE_VERIFY_FAILED' in str(e): msg += (' [this is a LOCAL TLS trust-store problem, not a bad price: your Python has no ' 'root certificates. Run Install Certificates.command (macOS python.org builds), ' 'or point SSL_CERT_FILE at a real CA bundle.]') return msg def re_derive_kcc20(feed, kascov_base=None, kas_usd=None): """Re-derive one L1 KCC20 feed's price from kascov itself. NETWORK CALL. Returns {'ok', 'basis', 'expected_basis', 'exact', 'published', 'rederived', 'price', 'published_price', 'drift_pct', 'usd_checked', 'checks', 'urls', 'error'} where checks is [{'name','ok','detail'}]. `ok` is True only if the tier AND both integers reproduce exactly. `kascov_base` overrides the origin. The DEFAULT is KCC20_PINNED_ORIGIN, not the feed's own `price_source_url` — the audited party does not get to choose where it is audited; its URL is checked against the pin instead. `kas_usd` — pass YOUR OWN KAS/USD to also check the `/USD` leg. Without it only the TOKEN/KAS rational is reproduced, `usd_checked` is False, and the caller must not report the dollar price as re-derived.""" info = kcc20_info(feed) out = {'ok': False, 'pair': (feed or {}).get('pair'), 'basis': None, 'expected_basis': None, 'exact': False, 'published': None, 'rederived': None, 'price': None, 'published_price': None, 'drift_pct': None, 'usd_checked': False, 'checks': [], 'urls': [], 'error': None} if info is None: out['error'] = 'not a kcc20-pool feed' return out if _KCC20_SELF_TEST_ERROR is not None: # a broken verifier refuses, never ✓ out['error'] = 'kcc20 self-test failed: ' + _KCC20_SELF_TEST_ERROR out['checks'].append({'name': 'self-test', 'ok': False, 'detail': out['error']}) return out out['basis'] = info['basis'] if info['unpriced_reason']: out['error'] = ('the oracle published this price with no provenance: %s' % info['unpriced_reason']) out['checks'].append({'name': 'provenance', 'ok': False, 'detail': out['error']}) return out origin = (str(kascov_base).rstrip('/') if kascov_base else None) or KCC20_PINNED_ORIGIN pools_url = '%s/data/%s/pools' % (origin, KCC20_NET) out['urls'].append(pools_url) try: tier = kcc20_tier_of(_kcc20_get(pools_url), info['market_covenant_id']) except (urllib.error.URLError, OSError, ValueError) as e: out['error'] = _kcc20_err(pools_url, e) return out if not tier['found']: out['error'] = 'kascov serves no pool with market_id %s' % info['market_covenant_id'] out['checks'].append({'name': 'market', 'ok': False, 'detail': out['error']}) return out out['expected_basis'] = tier['basis'] num, den = tier['num'], tier['den'] if tier['basis'] == 'last_verified_state': c_url = '%s/data/%s/token/%s/candles?bucket=1h&phase=pool' % (origin, KCC20_NET, info['token_covenant_id']) out['urls'].append(c_url) try: close = kcc20_candle_close(_kcc20_get(c_url)) except (urllib.error.URLError, OSError, ValueError) as e: out['error'] = _kcc20_err(c_url, e) return out if close: num, den = close['num'], close['den'] # the source URL must sit on the PINNED origin, must be the one that actually # carries this price for this tier, and must name THIS token. The origin half # is the load-bearing one: without it the feed picks the server that confirms # it. A URL pointing at another token's candles looks plausible and re-derives # to a different number, which is the other half. src = info['price_source_url'] or '' origin_ok = _kcc20_origin(src) == origin url_ok = origin_ok and (('candles' in src and str(info['token_covenant_id']) in src) if tier['basis'] == 'last_verified_state' else '/pools' in src) out['checks'].append({'name': 'source url', 'ok': url_ok, 'detail': src if origin_ok else '%s — NOT the pinned origin %s; the feed does not choose where it is checked' % (src, origin)}) out['checks'].append({'name': 'basis', 'ok': info['basis'] == tier['basis'], 'detail': 'kascov %s spot -> tier is "%s"; feed says "%s"' % ('publishes' if tier['basis'] == 'spot' else 'withholds', tier['basis'], info['basis'])}) f_num, f_den = _kcc20_int(info['num_sompi']), _kcc20_int(info['den']) out['published'] = None if f_num is None or f_den is None else {'num': f_num, 'den': f_den} out['rederived'] = None if num is None or den is None else {'num': num, 'den': den} exact = (num is not None and den is not None and f_num is not None and f_den is not None and num == f_num and den == f_den) # a different reduction of the same rational is honest, but it is not the # same two integers and must not read as "identical" same_value = (not exact and None not in (num, den, f_num, f_den) and den > 0 and f_den > 0 and num * f_den == f_num * den) out['exact'] = exact out['checks'].append({'name': 'exact rational', 'ok': exact or same_value, 'detail': 'mine %s/%s feed %s/%s -> %s' % ( num, den, f_num, f_den, 'IDENTICAL' if exact else 'equal value, different reduction' if same_value else 'MISMATCH')}) # the signed float against the rational it claims. On the /KAS leg this is a # 12-sample TWAP median, so it is allowed to lag an instantaneous rational — # reported as a measurement inside a stated band, never as equality. if num is not None and den is not None and den > 0: out['price'] = num / den / 1e8 px = feed.get('median', feed.get('price')) if isinstance(px, (int, float)) and info['leg'] == 'KAS' and out['price'] > 0: out['published_price'] = px out['drift_pct'] = (px - out['price']) / out['price'] * 100.0 out['checks'].append({'name': 'signed price', 'ok': abs(out['drift_pct']) <= 10.0, 'detail': 'feed %r vs rational %r -> %.3f%% (TWAP window)' % (px, out['price'], out['drift_pct'])}) # THE /USD LEG. Everything above this line is a property of the TOKEN/KAS # rational; nothing in it touches the dollar price, which is that rational # x the oracle's KAS/USD. Left unchecked, a wrong KAS/USD multiplier makes # every USD leg wrong by that factor and this still returns ok=True. So: # check it against a KAS/USD the CALLER supplies (never one taken from the # same envelope — that would be the feed grading itself), and when the # caller supplies none, say plainly that the dollar half was not checked. if isinstance(px, (int, float)) and info['leg'] == 'USD' and out['price'] > 0: out['published_price'] = px ku = float(kas_usd) if kas_usd not in (None, 0) else None if ku and ku > 0: implied = px / out['price'] # dollars per KAS the feed used d = (implied - ku) / ku * 100.0 out['usd_checked'] = True out['checks'].append({'name': 'USD leg', 'ok': abs(d) <= 1.0, 'detail': "feed's KAS/USD works out to $%.6f · yours $%.6f -> %.3f%%" % (implied, ku, d)}) else: out['checks'].append({'name': 'USD leg', 'ok': True, 'detail': 'NOT CHECKED — only the TOKEN/KAS rational was reproduced. ' 'Pass kas_usd (your own KAS/USD) to check the dollar multiplier too'}) out['ok'] = all(c['ok'] for c in out['checks']) if not out['ok']: out['error'] = 'the published price did not re-derive from kascov' return out def _kcc20_self_test(): """Pure self-test of the tier logic — no network. The fixtures are trimmed from live mainnet documents. Breaking the tier decision breaks the most dangerous thing on the wire, so it is checked at import like the crypto core.""" t1 = kcc20_tier_of({'pools': [{'market_id': 'f10e', 'market': { 'spot_num_sompi': 3029513000000, 'spot_den': 73048363}}]}, 'f10e') if t1['basis'] != 'spot' or t1['num'] != 3029513000000 or t1['den'] != 73048363: return 'spot tier misread' withheld = {'pools': [{'market_id': '10f2', 'market': {'reserve_sompi': 45220170000000}}]} t2 = kcc20_tier_of(withheld, '10f2') if t2['basis'] != 'last_verified_state' or t2['num'] is not None: return 'a withheld spot must fall through to tier 2, never to a reserve ratio' if kcc20_tier_of(withheld, 'nope')['found']: return 'an unknown market_id must not match a row' cl = kcc20_candle_close({'candles': [ {'t': 1, 'carried': False, 'close': {'quote_sompi': '1', 'base_amount': '2'}}, {'t': 2, 'carried': False, 'close': {'quote_sompi': '12995162000000', 'base_amount': '353889693'}}, {'t': 3, 'carried': True, 'close': {'quote_sompi': '12995162000000', 'base_amount': '353889693'}}]}) if not cl or cl['t'] != 2 or cl['num'] != 12995162000000: return 'carried candles must be skipped' # the origin pin: a price_source_url on any other host must not resolve to # the audit origin, because that is how a forged feed picks its own auditor if _kcc20_origin('http://127.0.0.1:8200/data/mainnet/pools') == KCC20_PINNED_ORIGIN: return 'origin comparison is broken' if KCC20_PINNED_ORIGIN != 'https://kascov.io': return 'the kascov origin pin was changed' return None # ── independently pinned committee verification ───────────────────────────── def _decode_pinned_committee(committee): if not isinstance(committee, dict): return None, None, 'committee pin: missing committee' need = committee.get('threshold') num_nodes = committee.get('num_nodes') signers = committee.get('signers') if type(need) is not int or need <= 0 or not isinstance(signers, list) or not signers: return None, None, 'committee pin: empty or zero-threshold committee' if type(num_nodes) is not int or num_nodes != len(signers): return None, None, 'committee pin: num_nodes does not match signer count' pinned = set() for encoded in signers: pk = _hex_bytes(encoded) if pk is None or len(pk) != 32 or _lift_x(int.from_bytes(pk, 'big')) is None: return None, None, 'committee pin: malformed signer key' if pk in pinned: return None, None, 'committee pin: duplicate signer key' pinned.add(pk) if need > len(pinned): return None, None, 'committee pin: threshold exceeds distinct signer count' return need, pinned, None def _count_pinned_valid(results, pinned): valid = set() for row in results if isinstance(results, list) else []: if not isinstance(row, dict) or row.get('ok') is not True: continue pk = _hex_bytes(row.get('signer')) if pk is not None and len(pk) == 32 and pk in pinned: valid.add(pk) return len(valid) def verify_with_committee(feed, committee): """Verify a feed against an independently supplied committee. Requires the committee threshold of distinct, valid signatures from decoded pinned keys in both the message and cov/v2 domains. A pinned key merely appearing beside an invalid signature is never credited. """ r = verify_feed(feed) need, pinned, error = _decode_pinned_committee(committee) if error: r = dict(r); r['ok'] = False; r['error'] = error return r if not r.get('ok'): return r message_valid = _count_pinned_valid(r.get('results'), pinned) if message_valid < need: r = dict(r); r['ok'] = False; r['pinned_valid'] = message_valid r['error'] = ('committee pin: fewer than committee-threshold valid message ' 'signatures from pinned keys') return r cov = r.get('covenant') or {'present': False} covenant_valid = None if cov.get('present'): covenant_valid = _count_pinned_valid(cov.get('results'), pinned) if covenant_valid < need: r = dict(r); r['ok'] = False; r['pinned_valid'] = message_valid r['pinned_covenant_valid'] = covenant_valid r['error'] = ('committee pin: fewer than committee-threshold valid covenant ' 'signatures from pinned keys') return r r = dict(r); r['pinned_valid'] = message_valid r['pinned_covenant_valid'] = covenant_valid return r # ── HTTP client ───────────────────────────────────────────────────────────── class NoSuchFeed(Exception): def __init__(self, pair): super().__init__('kaspulse: no such feed: %s' % pair) self.pair = pair # Public hosted oracle. Applications may explicitly configure another base URL. DEFAULT_BASE = 'https://pulse.kascov.io' class Kaspulse: def __init__(self, base_url=DEFAULT_BASE): self.base_url = str(base_url).rstrip('/') def _get(self, path): req = urllib.request.Request(self.base_url + path, headers={'Accept': 'application/json'}) with urllib.request.urlopen(req, timeout=10) as resp: return json.load(resp) def feeds(self): """Light catalog: {round, timestamp, count, feeds:[{pair, price, ...}]}. This is the endpoint dashboards should poll.""" return self._get('/v1/feeds') def committee(self): """Candidate committee artifact from the feed origin. Compare or persist its keys and threshold through an independent channel before treating it as a pin for verify_with_committee(). """ return self._get('/v1/committee') def feed(self, pair): """One full FeedObj (price, sources, signatures, history). pair like 'KAS/USD' or 'KAS-USD', case-insensitive. Unknown pair → NoSuchFeed.""" try: return self._get('/v1/feed/' + str(pair).replace('/', '-')) except urllib.error.HTTPError as e: if e.code == 404: raise NoSuchFeed(pair) from None raise def verify_covenant(self, feed): """Verify the feed's BOUND `kaspulse/cov/v2` covenant attestation — the preimage an on-chain price gate consumes. REBUILDS the blob from the feed's own pair/mant/expo/signed_round/signed_ts and demands covenant.preimage back byte-for-byte before it looks at a signature. Pure, no network. 'present': False = this feed carries no covenant, which is not a failure.""" return verify_covenant(feed) def verify_feed(self, feed): """See module-level verify_feed().""" return verify_feed(feed) def verify_with_committee(self, feed, committee): """See module-level verify_with_committee().""" return verify_with_committee(feed, committee) def checked_value(self, feed, max_age_s=30): """See module-level checked_value().""" return checked_value(feed, max_age_s) # ── L1 KCC20 (Kaspa L1 AMM pools, priced through kascov) ──────────────── def kcc20_feeds(self): """The L1 rows of the light catalog: kind == "kcc20-pool", each carrying `basis` and `last_trade_age_s` so a dashboard never has to open the full envelope to know which tier priced a feed and how old the fill is.""" return [f for f in (self.feeds().get('feeds') or []) if f.get('kind') == 'kcc20-pool'] def kcc20_census(self): """The L1 census block from /v1/feed: how many markets kascov indexes, how many kaspulse prices, and EVERY market it refused with the reason. The refusals are the point — a market is never dropped silently.""" return self._get('/v1/feed').get('kcc20') def kcc20_info(self, feed): """See module-level kcc20_info().""" return kcc20_info(feed) def re_derive_kcc20(self, feed, kascov_base=None, kas_usd=None): """See module-level re_derive_kcc20(). NETWORK CALL. The kascov origin is PINNED, not read from the feed; pass kas_usd to also check the /USD leg's dollar multiplier.""" return re_derive_kcc20(feed, kascov_base, kas_usd) # ── mandatory import-time self-test: refuse to run with a broken core ─────── def _self_test(): abc = blake2b256(b'abc').hex() if abc != 'bddd813c634239723171ef3fee98579b94964e3bb1cb3e427262c8c068d52319': return 'blake2b-256(b"abc") known-answer mismatch' # BIP340 official test vector 0 pk = bytes.fromhex('f9308a019258c31049344f85f89d5229b531c845836f99b08601f113bce036f9') msg = bytes(32) sig = bytes.fromhex('e907831f80848d1069a5371b402410364bdf1c5f8307b0084c55f1ce2dca821525f66a4a85ea8b71e482a74f382d2ce5ebeee8fdb2172f477df4900d310536c0') if not bip340_verify(pk, msg, sig): return 'BIP340 test vector 0 did not verify' bad = bytearray(sig) bad[63] ^= 0x01 if bip340_verify(pk, msg, bytes(bad)): return 'corrupted BIP340 signature verified (core is broken)' # cov/v2 known-answer: KAS/USD, mant 290000000, expo -10, round 4242, # ts 1756000000. Pinned hex — this is the byte layout the on-chain script # slices with OpSubstr, so a builder that drifts by one byte must not be # allowed to print a ✓ anywhere. cov = cov_v2_preimage('KAS/USD', 290000000, -10, 4242, 1756000000) if cov is None or cov.hex() != ('6b617370756c73652f636f762f7632b84ad8389aa2ebb0f6' '00000000000010920000000068aa6f00800c4911'): return 'cov/v2 preimage known-answer mismatch (got %s)' % (cov.hex() if cov else 'None') # the field binding must cover ALL FIVE signed fields. Each candidate below # differs from the reference in exactly one of them and must fail to bind. bref = {'pair': 'KAS/USD', 'mant': 290000000, 'expo': -10, 'signed_ts': 1756000000, 'signed_round': 4242} bmsg = parse_signed_message('kaspulse/v2|KAS/USD|290000000|-10|1756000000|4242') if not _message_binds(bmsg, bref): return 'field binding rejected a matching feed' for k in ('pair', 'mant', 'expo', 'signed_ts', 'signed_round'): bad_feed = dict(bref) bad_feed[k] = 'BTC/USD' if k == 'pair' else bref[k] + 1 if _message_binds(bmsg, bad_feed): return 'field binding ignores ' + k # the two swaps that broke the OLD domain must change the BYTES, or the # byte-for-byte check in verify_covenant is decoration if cov_v2_preimage('BTC/USD', 290000000, -10, 4242, 1756000000) == cov: return 'cov/v2 preimage does not bind the pair' if cov_v2_preimage('KAS/USD', 290000000, -9, 4242, 1756000000) == cov: return 'cov/v2 preimage does not bind the exponent' # and the shapes that must refuse to build at all if cov_v2_preimage('KAS/USD', 0, -10, 4242, 1756000000) is not None: return 'cov/v2 built a ZERO mantissa (the blake2b(empty) hole)' if cov_v2_preimage('KAS/USD', 290000000, 200, 4242, 1756000000) is not None: return 'cov/v2 built an expo that does not fit one byte' # ONE key, listed THREE times with its one genuine signature, must never # count as three. This is the off-chain threshold collapse: a single # compromised committee key plus control of the JSON forges any price. The # dedupe is one set() in _count_distinct_valid and is exactly the kind of # line that gets refactored away, so it is pinned here at import time. dup_n = _count_distinct_valid([pk.hex()] * 3, [sig.hex()] * 3, msg, []) if dup_n != 1: return 'one signer listed 3x counted as %d (threshold collapse)' % dup_n # and hex case must not mint a second signer cased_n = _count_distinct_valid([pk.hex(), pk.hex().upper()], [sig.hex()] * 2, msg, []) if cased_n != 1: return 'a re-cased pubkey counted as a second signer' # Committee membership must be a property of a VALID signature, not a # separate label-overlap check. This is the exact old bypass in miniature: # an attacker signature verifies while a pinned key is present with junk. pinned_key = bytes.fromhex('79be667ef9dcbbac55a06295ce870b07029bfcdb2dce28d959f2815b16f81798') need, pinned, error = _decode_pinned_committee({ 'threshold': 1, 'num_nodes': 1, 'signers': [pinned_key.hex()]}) if error or need != 1: return 'valid committee pin rejected: ' + str(error) rows = [{'signer': pk.hex(), 'ok': True}, {'signer': pinned_key.hex(), 'ok': False}] if _count_pinned_valid(rows, pinned) != 0: return 'invalid pinned signer label was credited as a valid committee signature' for malformed in ( {'threshold': 0, 'num_nodes': 1, 'signers': [pinned_key.hex()]}, {'threshold': 2, 'num_nodes': 1, 'signers': [pinned_key.hex()]}, {'threshold': 1, 'num_nodes': 2, 'signers': [pinned_key.hex()]}, {'threshold': 1, 'num_nodes': 1, 'signers': ['00']}, {'threshold': 1, 'num_nodes': 2, 'signers': [pinned_key.hex(), pinned_key.hex().upper()]}, ): if _decode_pinned_committee(malformed)[2] is None: return 'malformed committee pin was accepted' now_s = 1800000000.0 if _freshness_error(now_s, 0, now_s) is not None: return 'current signed timestamp failed a zero-age check' if 'future' not in (_freshness_error(now_s + 31, 60, now_s) or ''): return 'timestamp more than 30 seconds in the future was accepted' if 'old' not in (_freshness_error(now_s - 31, 30, now_s) or ''): return 'timestamp older than max_age_s was accepted' for bad_age in (-1, float('nan'), float('inf'), float('-inf'), '30', None, True): if _max_age_error(bad_age) is None: return 'invalid max_age_s was accepted: %r' % (bad_age,) return None _SELF_TEST_ERROR = _self_test() if _SELF_TEST_ERROR is not None: raise RuntimeError('kaspulse verifier self-test failed: ' + _SELF_TEST_ERROR) # The KCC20 tier logic gets the same treatment. It does NOT raise on import: # a broken tier decision must not take down signature verification, which is a # separate concern and still perfectly sound. re_derive_kcc20 refuses instead. _KCC20_SELF_TEST_ERROR = _kcc20_self_test() if _KCC20_SELF_TEST_ERROR is not None: sys.stderr.write('kaspulse: kcc20 self-test failed, L1 re-derivation disabled: %s\n' % _KCC20_SELF_TEST_ERROR) # ── CLI: python3 kaspulse.py verify KAS/USD [base] ────────────────────────── def _kcc20_main(base): """python3 kaspulse.py kcc20 [base] — signature-verify AND re-derive every L1 KCC20 price straight from kascov. The same two checks the Rust `verify` binary runs, minus the gate replay (see the KCC20 block for why).""" k = Kaspulse(base) if _KCC20_SELF_TEST_ERROR is not None: print('✗ kcc20 self-test failed: %s' % _KCC20_SELF_TEST_ERROR, file=sys.stderr) return 1 print('kaspulse kcc20 — re-deriving every L1 price from kascov (%s)' % base) try: census = k.kcc20_census() rows = k.kcc20_feeds() except (urllib.error.URLError, OSError) as e: print('✗ kaspulse: cannot reach %s: %s' % (base, e), file=sys.stderr) return 1 if census: print(' kascov %s · %s graduated pools of %s markets · priced %s · refused %s ' '(each with a published reason)' % (census.get('source'), census.get('pools_total'), census.get('markets_total'), census.get('priced'), census.get('skipped_count'))) if not rows: print(' no kcc20-pool feeds in this envelope') return 0 # An INDEPENDENT KAS/USD for the /USD legs. Everything re_derive_kcc20 checks # is a property of the TOKEN/KAS rational, so without this a wrong KAS/USD # multiplier leaves every dollar price wrong and every leg still ✓. Kraken # direct, over the stdlib urllib already imported — no new dependency, and # deliberately NOT the oracle's own KAS/USD, which would be the thing under # audit grading itself. Unreachable is reported, never assumed. kas_usd = None try: kj = _kcc20_get('https://api.kraken.com/0/public/Ticker?pair=KASUSD') v = float(list(kj.get('result', {}).values())[0]['c'][0]) if v > 0: kas_usd = v except (urllib.error.URLError, OSError, ValueError, KeyError, IndexError): pass print(' my own KAS/USD for the /USD legs: $%s (Kraken, fetched by this process — not the oracle\'s)' % kas_usd if kas_usd else ' /USD legs will NOT be dollar-checked — could not reach Kraken for an independent KAS/USD') bad = 0 for row in rows: feed = k.feed(row['pair']) sig = k.verify_feed(feed) r = k.re_derive_kcc20(feed, kas_usd=kas_usd) info = k.kcc20_info(feed) or {} ok = r['ok'] and (sig['ok'] or feed.get('halted')) if not ok: bad += 1 print('\n %s %s %s' % ('✓' if ok else '✗', feed.get('pair'), info.get('display_name') or '')) print(' signatures %d/%d verify (threshold %d)%s%s' % (sig['valid'], len(sig['results']), sig['threshold'], ' · feed HALTED by its own circuit breaker' if feed.get('halted') else '', '' if sig['ok'] or sig.get('halted') or sig.get('depegged') else ' · ' + str(sig.get('error')))) print(' basis %s · last trade %ss ago · anchor %s · gates %s/%s admitted, %s bracket-rejected' % (info.get('basis'), info.get('last_trade_age_s'), str(info.get('anchor_txid'))[:16], info.get('admitted'), info.get('gate_window'), info.get('rejected_bracket'))) for c in r['checks']: print(' %s %-15s %s' % ('✓' if c['ok'] else '✗', c['name'], c['detail'])) # a re-derivation that never got far enough to produce a check (kascov # unreachable, TLS refused) must still say WHY — an empty ✗ is useless if not r['ok'] and r.get('error'): print(' ✗ %-15s %s' % ('error', r['error'])) for u in r['urls']: print(' curl %s' % u) if bad == 0: print('\n✓ ALL %d L1 KCC20 LEGS RE-DERIVED — every exact integer rational reproduced from kascov%s.' '\n Not independence: this re-fetches kascov and compares. A bug inside kascov reproduces here too.' % (len(rows), ", and every /USD leg checked against this process's own Kraken KAS/USD" if kas_usd else "; the /USD legs' dollar multiplier was NOT checked (no independent KAS/USD)")) return 0 print('\n✗ %d of %d L1 KCC20 legs DID NOT re-derive. Do not consume this feed.' % (bad, len(rows))) return 1 def _main(argv): if argv and argv[0] == 'kcc20': return _kcc20_main(argv[1] if len(argv) > 1 else DEFAULT_BASE) if len(argv) < 2 or argv[0] != 'verify': print('usage: python3 kaspulse.py verify KAS/USD [base]', file=sys.stderr) print(' python3 kaspulse.py kcc20 [base] ' '# re-derive every L1 KCC20 price from kascov', file=sys.stderr) return 2 pair = argv[1].replace('-', '/') base = argv[2] if len(argv) > 2 else DEFAULT_BASE k = Kaspulse(base) try: feed = k.feed(pair) except NoSuchFeed as e: print('✗ %s' % e, file=sys.stderr) return 1 except (urllib.error.URLError, OSError) as e: print('✗ kaspulse: cannot reach %s: %s' % (base, e), file=sys.stderr) return 1 r = verify_feed(feed) print('%s %s signed_round %s' % (feed.get('pair'), base, feed.get('signed_round', '?'))) for i, node in enumerate(r['results']): print(' node %d %s %s' % (i, '✓' if node['ok'] else '✗', node['signer'])) print(' bound=%s (signed message fields == JSON fields)' % str(r['bound']).lower()) # ── the covenant domain. A feed with no covenant is fine; a feed whose # covenant does NOT rebuild is the withdrawn hole, and must never come # out of here as a green ✓ (verify_feed already folded it into r['ok']). c = r['covenant'] if not c['present']: print(' covenant — none published on this feed (kaspulse/cov/v2)') else: print(" covenant kaspulse/cov/v2 preimage rebuilt from THIS feed's fields:") if c['expected'] and c['published']: print(' expected %s' % c['expected']) print(' published %s' % c['published']) print(' byte-for-byte match: %s%s' % ('YES' if c['matches'] else 'NO', (' · %d/%d signatures over blake2b256(preimage) (threshold %d)' % (c['valid'], len(c['results']), c['threshold'])) if c['matches'] else '')) if not c['ok']: print(' ✗ COVENANT %s FAILED — %s' % ('SIGNATURES' if c['matches'] else 'DOMAIN', c.get('error'))) sigs_ok = r['bound'] and r['threshold'] > 0 and r['valid'] >= r['threshold'] cov_fatal = c['present'] and not c['ok'] # A HALTED feed is the breaker WORKING, not a forgery: its signatures are # valid and the committee is honest, and the price it carries is the held one # the breaker is suppressing. Printing a bare '✓ VALID … price = X' for it — # which this path did until 2026-08-25 — hands a consumer exactly the number # they must not use. It is its own third verdict, never a green ✓. if sigs_ok and not cov_fatal and (r.get('halted') or r.get('depegged')): print('⚠ NOT SAFE TO CONSUME — %s' % r['error']) print(' the %d/%d committee signatures DO verify (threshold %d)%s.' % (r['valid'], len(r['results']), r['threshold'], (', and the covenant is BOUND to %s @ expo %s round %s' % (feed.get('pair'), feed.get('expo'), feed.get('signed_round'))) if c['present'] else '')) print(' the price this feed carries is the HELD one its own circuit breaker ' 'is suppressing. Do not consume it.') return 1 if r['ok']: px = int(feed['mant']) * 10.0 ** int(feed['expo']) print('✓ VALID — %d/%d signatures verify (threshold %d), price = %s%s' % (r['valid'], len(r['results']), r['threshold'], px, (', covenant BOUND to %s @ expo %s round %s' % (feed.get('pair'), feed.get('expo'), feed.get('signed_round'))) if c['present'] else '')) return 0 print('✗ INVALID — %s' % r.get('error', 'verification failed')) return 1 if __name__ == '__main__': sys.exit(_main(sys.argv[1:]))