#!/usr/bin/env python3 """implementare-integritate.py: an INDEPENDENT implementation, written from the rules and not from the JavaScript code, in another language, with nothing but the Python standard library. It exists to show that the conformance vectors are not tied to one implementation: the same runner scores it, and it passes every vector whose rules it claims. It claims little, on purpose, and says so through --capabilities: envelope envelope.form, envelope.integrity (it cannot check an ML-DSA signature or read a chain: vectors that exercise those rules are reported by the runner as NOT COVERED, never as passed) dataset-record dataset.form, dataset.commitment, dataset.field, dataset.inclusion (RFC 9162 inclusion, salted field commitments) device-reputation reputation.form, reputation.figures (2026-10-20: the formula aere-device-reputation/1 of the device operation component, computed again from the facts and compared with the figures stated; written from the formula in words in that component's README, not from its code: the windows are found by bisection over their starts, not by division, the gaps by one pass over the windows, and the comparison is of canonical texts written by this file, never of Python values, in which True == 1. The form of the facts follows the members and bounds that README names and the patterns the published form gives; the operation log itself carries ML-DSA-65 signatures, which this file cannot check, so its vectors are NOT COVERED) python3 implementare-integritate.py --capabilities python3 implementare-integritate.py python3 implementare-integritate.py --batch Limits of this implementation: numbers are written the way JavaScript writes them only for integers and for the common decimal floats (no exponent forms); object keys are sorted by code point, which equals JavaScript's order for keys in the Basic Multilingual Plane. Files are read with Python's json module: a file nested deeper than its recursion limit is not read (an error, which the runner scores BROKEN, never a verdict), and the non-JSON words NaN and Infinity, which JSON.parse refuses, are read. It never opens a connection: --online is not claimed. """ import bisect import hashlib import json import re import sys # a whole number of 5,000 digits is read, as JavaScript reads it (to a number it then judges), instead of stopping the reader if hasattr(sys, 'set_int_max_str_digits'): sys.set_int_max_str_digits(0) LEVELS = ['integrity', 'authenticity', 'anchoring', 'finality'] H64 = re.compile(r'^0x[0-9a-fA-F]{64}$') H64_LOW = re.compile(r'^0x[0-9a-f]{64}$') KIND_RECORD = 'aere-verifiable-record' def sha256_hex(b): return '0x' + hashlib.sha256(b).hexdigest() def js_number(x): if isinstance(x, bool): raise ValueError('not a number') if isinstance(x, int): return str(x) if x != x or x in (float('inf'), float('-inf')): raise ValueError('a number must be finite') if x == int(x) and abs(x) < 1e21: return str(int(x)) r = repr(x) if 'e' in r or 'E' in r: raise ValueError('exponent form: not reproduced by this implementation') return r def js_stringify(x): """JSON.stringify(x) for parsed JSON: insertion order of keys, no whitespace.""" if x is None: return 'null' if x is True: return 'true' if x is False: return 'false' if isinstance(x, str): return json.dumps(x, ensure_ascii=False) if isinstance(x, (int, float)): return js_number(x) if isinstance(x, list): return '[' + ','.join(js_stringify(v) for v in x) + ']' if isinstance(x, dict): return '{' + ','.join(json.dumps(k, ensure_ascii=False) + ':' + js_stringify(v) for k, v in x.items()) + '}' raise ValueError('unsupported value') def canonical(x): """Canonical JSON of the dataset format: keys sorted at every depth, no whitespace.""" if isinstance(x, list): return '[' + ','.join(canonical(v) for v in x) + ']' if isinstance(x, dict): return '{' + ','.join(json.dumps(k, ensure_ascii=False) + ':' + canonical(x[k]) for k in sorted(x.keys())) + '}' return js_stringify(x) def verdict(artifact, failed_by_level, reasons, passed, partial=None): """The one verdict form: four levels, each PASSED, FAILED or NOT_MEASURED with its reason.""" levels = [] for l in LEVELS: if failed_by_level.get(l): levels.append({'level': l, 'state': 'FAILED', 'rules': sorted(set(failed_by_level[l])), 'reason': '; '.join(reasons)}) elif partial and l in partial: levels.append({'level': l, 'state': 'NOT_MEASURED', 'rules': [], 'cause': partial[l][0], 'reason': partial[l][1]}) elif l in passed: levels.append({'level': l, 'state': 'PASSED', 'rules': [], 'reason': passed[l]}) else: levels.append({'level': l, 'state': 'NOT_MEASURED', 'rules': [], 'cause': 'not-judged', 'reason': 'this implementation does not judge this level'}) failed = sorted({r for x in levels for r in x['rules']}) is_partial = any(x['state'] == 'NOT_MEASURED' and x.get('cause') in ('claim-not-checked', 'unmeasurable') for x in levels) v = 'INVALID' if failed else ('PARTIAL' if is_partial or levels[0]['state'] != 'PASSED' else 'VALID') return {'v': 1, 'kind': 'aere-conformance-verdict', 'artifact': artifact, 'verdict': v, 'levels': levels, 'failed': failed} def judge_envelope(inp): env = inp.get('envelope') if isinstance(inp.get('raw'), str): # the text of a file, in place of the envelope: it may not be JSON at all try: env = json.loads(inp['raw']) except ValueError: return verdict('envelope', {'integrity': ['envelope.form']}, ['the file is not JSON'], {}) ok_form = isinstance(env, dict) and isinstance(env.get('statement'), dict) and isinstance(env.get('statementHash'), str) and bool(H64.match(env['statementHash'])) if not ok_form: return verdict('envelope', {'integrity': ['envelope.form']}, ['the envelope lacks a statement (object) and a statementHash (0x + 64 hex)'], {}) h = sha256_hex(js_stringify(env['statement']).encode('utf-8')) if h != env['statementHash'].lower(): return verdict('envelope', {'integrity': ['envelope.integrity']}, ['sha256 of the statement text is ' + h[:18] + '.., not the statementHash'], {}) partial = {} if env.get('signature'): partial['authenticity'] = ('unmeasurable', 'the envelope carries a signature this implementation cannot check') if env.get('notarization'): why = ('claim-not-checked', 'the envelope claims a notarization; this implementation never reads a chain') partial['anchoring'] = why partial['finality'] = why return verdict('envelope', {}, [], {'integrity': 'form and statement hash hold'}, partial) def node_hash(a, b): return hashlib.sha256(b'\x01' + a + b).digest() def unhex(s): if not isinstance(s, str) or not H64_LOW.match(s): raise ValueError('not a 32-byte hash') return bytes.fromhex(s[2:]) def is_int(x): return type(x) is int def verify_inclusion(leaf, index, tree_size, path, root): """RFC 9162, 2.1.3.2.""" try: if not (is_int(index) and is_int(tree_size) and 0 <= index < tree_size and isinstance(path, list)): return False fn, sn, r = index, tree_size - 1, unhex(leaf) for p in path: pb = unhex(p) if sn == 0: return False if (fn & 1) or fn == sn: r = node_hash(pb, r) if not (fn & 1): while not (fn & 1) and fn != 0: fn >>= 1 sn >>= 1 else: r = node_hash(r, pb) fn >>= 1 sn >>= 1 return sn == 0 and '0x' + r.hex() == str(root) except ValueError: return False def judge_record(inp): rec, com = inp.get('record'), inp.get('commitment') failed, reasons = [], [] if not isinstance(rec, dict) or rec.get('kind') != KIND_RECORD or not isinstance(com, dict): return verdict('dataset-record', {'integrity': ['dataset.form']}, ['not a ' + KIND_RECORD], {}) if not (rec.get('commitmentHash') == sha256_hex(canonical(com).encode('utf-8')) and rec.get('version') == com.get('version')): failed.append('dataset.commitment') reasons.append('the record names another commitment') def row(): c, disclosed = rec.get('c'), rec.get('disclosed') if not (isinstance(c, (dict, list)) and isinstance(disclosed, (dict, list))): failed.append('dataset.form') reasons.append('malformed row') return for k, d in (disclosed.items() if isinstance(disclosed, dict) else enumerate(disclosed)): if not (isinstance(d, dict) and isinstance(d.get('salt'), str)): failed.append('dataset.field') reasons.append('disclosed field ' + str(k) + ' without a salt') return want = sha256_hex((d['salt'] + '|' + str(k) + '|' + canonical(d.get('value'))).encode('utf-8')) if not (isinstance(c, dict) and c.get(k) == want): failed.append('dataset.field') reasons.append('the disclosed value of "' + str(k) + '" does not match its commitment') return leaf = sha256_hex(b'\x00' + canonical({'id': rec.get('id'), 'c': c}).encode('utf-8')) lp = rec.get('logProof') if not (isinstance(lp, dict) and verify_inclusion(leaf, lp.get('index'), com.get('size'), lp.get('path'), com.get('logRoot')) and lp.get('index') == rec.get('i')): failed.append('dataset.inclusion') reasons.append('the record is not in the log of version ' + str(com.get('version'))) row() passed = {} if failed else {'integrity': 'the record opens its commitments and is in the log'} return verdict('dataset-record', {'integrity': failed} if failed else {}, reasons, passed) # ---------------------------------------------------------------- device-reputation (2026-10-20) # The formula, in the words of the device operation README ("The reputation formula"): windows of W = windowS * 1000 ms over [from, to); # a counted record covers by the device's clock the window its `at` falls in, and is witnessed in the window its `lo` falls in when its whole # interval [lo, hi] lies in that window; availability, per mille rounded down, the runs of windows not covered by the clock, the proofs of # equivocation and the retirements copied, the age from the first counted record and from the lowest witnessed upper bound, the evidence # of every window covered either way, and the hash of the canonical facts. Integers only. REP_KIND, REP_FORMULA, FACTS_KIND = 'aere-device-reputation', 'aere-device-reputation/1', 'aere-device-operation-facts' FACT_MEMBERS = ('v', 'kind', 'verdict', 'rules', 'unjudged', 'manufacturer', 'keys', 'records', 'uncounted', 'equivocations', 'retirements', 'chains') REC_MEMBERS = ('seq', 'hash', 'at', 'lo', 'hi', 'state', 'counter') REP_STATES = ('operating', 'idle', 'fault', 'maintenance') REP_VERDICTS = ('VALID', 'PARTIAL', 'INVALID') SAFE_INT = 9007199254740991 # 2^53 - 1: what JavaScript holds exactly TIME_MAX = 4503599627370496 # 2^52 ms WINDOW_MAX_S = 366 * 86400 WINDOWS_MAX, RECORDS_MAX, KEYS_MAX, EQUIV_MAX, CHAINS_MAX, NAMES_MAX = 100000, 100000, 16, 4096, 64, 64 P_NAME = re.compile(r'[a-z][a-z0-9-]{0,63}') P_HASH = re.compile(r'0x[0-9a-f]{64}') P_DEVICE = re.compile(r'aere-device:[0-9a-f]{40}') P_AERE_ID = re.compile(r'aere-id:[0-9a-f]{40}') def whole(x, lo=0, hi=SAFE_INT): """The whole number JavaScript reads from this JSON value, or None. Python's json reads '7' as int and '7.0' or '7e0' as float; JavaScript reads all three as the number 7. A boolean is never a number (in Python True == 1).""" if isinstance(x, bool): return None if isinstance(x, float): if not x.is_integer(): return None x = int(x) if not isinstance(x, int): return None return x if lo <= x <= hi else None def text_of(x, pattern): return isinstance(x, str) and pattern.fullmatch(x) is not None def ascending(xs): return all(xs[k] < xs[k + 1] for k in range(len(xs) - 1)) def members(o, names): return isinstance(o, dict) and set(o) == set(names) def facts_problem(f): """None when the facts are of the form the formula reads; otherwise what is not.""" if not members(f, FACT_MEMBERS): return 'facts: not exactly ' + ', '.join(FACT_MEMBERS) if whole(f['v']) != 1 or f['kind'] != FACTS_KIND: return 'facts: not ' + FACTS_KIND + ' version 1' if not (isinstance(f['verdict'], str) and f['verdict'] in REP_VERDICTS): return 'facts: verdict' for k in ('rules', 'unjudged'): x = f[k] if not (isinstance(x, list) and len(x) <= NAMES_MAX and all(text_of(n, P_NAME) for n in x) and ascending(x)): return 'facts: ' + k + ' is a sorted list of distinct names' if not (f['manufacturer'] is None or text_of(f['manufacturer'], P_AERE_ID)): return 'facts: manufacturer' if not (isinstance(f['keys'], list) and len(f['keys']) <= KEYS_MAX and all(text_of(k, P_DEVICE) for k in f['keys'])): return 'facts: keys' rs = f['records'] if not (isinstance(rs, list) and len(rs) <= RECORDS_MAX): return 'facts: records' before = None for r in rs: if not members(r, REC_MEMBERS): return 'facts: a record is not exactly ' + ', '.join(REC_MEMBERS) seq, at, counter = whole(r['seq']), whole(r['at'], 0, TIME_MAX), whole(r['counter']) bounds_ok = all(r[b] is None or whole(r[b], 0, TIME_MAX) is not None for b in ('lo', 'hi')) if seq is None or at is None or counter is None or not bounds_ok or not text_of(r['hash'], P_HASH) or not (isinstance(r['state'], str) and r['state'] in REP_STATES): return 'facts: a record has a member of another form' if before is not None and not (before[0] < seq and before[1] <= at and before[2] <= counter): return 'facts: a record does not follow the one before it' before = (seq, at, counter) if whole(f['uncounted']) is None: return 'facts: uncounted' es = f['equivocations'] if not (isinstance(es, list) and len(es) <= EQUIV_MAX): return 'facts: equivocations' keys_e = [] for e in es: if not (members(e, ('seq', 'a', 'b')) and whole(e['seq']) is not None and text_of(e['a'], P_HASH) and text_of(e['b'], P_HASH) and e['a'] != e['b']): return 'facts: an equivocation is not { seq, a, b } with two different record hashes' keys_e.append((whole(e['seq']), e['a'], e['b'])) if not ascending(keys_e): return 'facts: equivocations are sorted by seq, a, b and distinct' ts = f['retirements'] if not (isinstance(ts, list) and len(ts) <= KEYS_MAX and all(members(t, ('deviceId', 'at')) and text_of(t['deviceId'], P_DEVICE) and (t['at'] is None or whole(t['at'], 0, TIME_MAX) is not None) for t in ts)): return 'facts: retirements' c = f['chains'] if not members(c, ('anchors', 'beacons')): return 'facts: chains' for k in ('anchors', 'beacons'): ids = c[k] if not (isinstance(ids, list) and len(ids) <= CHAINS_MAX and all(whole(i, 1) is not None for i in ids) and ascending([whole(i, 1) for i in ids])): return 'facts: chains.' + k return None def params_problem(p): if not members(p, ('from', 'to', 'windowS')): return 'params: not exactly from, to, windowS' a, b, w = whole(p['from']), whole(p['to']), whole(p['windowS'], 1, WINDOW_MAX_S) if a is None or b is None or not (a < b <= TIME_MAX): return 'params: from and to' if w is None: return 'params: windowS' if len(range(a, b, w * 1000)) > WINDOWS_MAX: return 'params: windows' return None def reputation_of(f, p): a, b, ws = whole(p['from']), whole(p['to']), whole(p['windowS']) span = ws * 1000 starts = list(range(a, b, span)) # window i begins at starts[i] n = len(starts) ends = starts[1:] + [b] # and ends at the next start, the last at `to` def window(t): # the window t falls in, or None outside [from, to) if not a <= t < b: return None return bisect.bisect_right(starts, t) - 1 recs = [{'seq': whole(r['seq']), 'at': whole(r['at']), 'lo': None if r['lo'] is None else whole(r['lo']), 'hi': None if r['hi'] is None else whole(r['hi']), 'state': r['state']} for r in f['records']] by_clock, by_witness = [None] * n, [None] * n in_range, states = 0, dict((s, 0) for s in REP_STATES) for r in recs: i = window(r['at']) if i is not None: in_range += 1 states[r['state']] += 1 by_clock[i] = r['seq'] if by_clock[i] is None else min(by_clock[i], r['seq']) if r['lo'] is not None and r['hi'] is not None and r['lo'] <= r['hi']: k = window(r['lo']) if k is not None and r['hi'] < ends[k]: by_witness[k] = r['seq'] if by_witness[k] is None else min(by_witness[k], r['seq']) runs, start = [], None for i in range(n + 1): empty = i < n and by_clock[i] is None if empty and start is None: start = i elif not empty and start is not None: runs.append([start, i - start]) start = None covered_c = sum(1 for x in by_clock if x is not None) covered_w = sum(1 for x in by_witness if x is not None) first = min(recs, key=lambda r: r['seq']) if recs else None his = [r['hi'] for r in recs if r['hi'] is not None] low_hi = min(his) if his else None return { 'v': 1, 'kind': REP_KIND, 'formula': REP_FORMULA, 'factsHash': sha256_hex(canonical(f).encode('utf-8')), 'params': {'from': a, 'to': b, 'windowS': ws}, 'verdict': f['verdict'], 'rules': f['rules'], 'unjudged': f['unjudged'], 'manufacturer': f['manufacturer'], 'keys': f['keys'], 'chains': f['chains'], 'records': {'inRange': in_range, 'byState': states, 'uncounted': f['uncounted']}, 'availability': {'windows': n, 'claimed': covered_c, 'witnessed': covered_w, 'claimedPerMille': (1000 * covered_c) // n, 'witnessedPerMille': (1000 * covered_w) // n}, 'gaps': {'count': len(runs), 'longestWindows': max([r[1] for r in runs] or [0]), 'list': runs}, 'equivocations': {'count': len(f['equivocations']), 'list': [[e['seq'], e['a'], e['b']] for e in f['equivocations']]}, 'retirements': {'count': len(f['retirements']), 'list': [[t['deviceId'], t['at']] for t in f['retirements']]}, 'age': {'claimedS': (b - first['at']) // 1000 if first is not None and first['at'] < b else None, 'witnessedS': (b - low_hi) // 1000 if low_hi is not None and low_hi < b else None}, 'evidence': {'windows': [[i, by_clock[i], by_witness[i]] for i in range(n) if by_clock[i] is not None or by_witness[i] is not None]}, } def judge_reputation(inp): if not members(inp, ('facts', 'params', 'claimed')): return verdict('device-reputation', {'integrity': ['reputation.form']}, ['the input is not exactly { facts, params, claimed }'], {}) why = facts_problem(inp['facts']) or params_problem(inp['params']) if why: return verdict('device-reputation', {'integrity': ['reputation.form']}, [why], {}) try: same = canonical(inp['claimed']) == canonical(reputation_of(inp['facts'], inp['params'])) except (ValueError, RecursionError): same = False # a claim this file cannot write as JavaScript writes it (an exponent form, Infinity) is not the formula's text if not same: return verdict('device-reputation', {'integrity': ['reputation.figures']}, ['the stated reputation is not what the formula gives for these facts and parameters'], {}) return verdict('device-reputation', {}, [], {'integrity': 'the stated reputation is what the formula ' + REP_FORMULA + ' gives'}) JUDGES = {'envelope': judge_envelope, 'dataset-record': judge_record, 'device-reputation': judge_reputation} CAPABILITIES = {'v': 1, 'kind': 'aere-conformance-capabilities', 'name': 'independent integrity checker (Python, standard library)', 'artifacts': sorted(JUDGES), 'rules': ['dataset.commitment', 'dataset.field', 'dataset.form', 'dataset.inclusion', 'envelope.form', 'envelope.integrity', 'reputation.figures', 'reputation.form'], 'online': False, 'batch': True} def one(path): try: with open(path, encoding='utf-8') as f: j = json.load(f) if not isinstance(j, dict) or j.get('artifact') not in JUDGES or not isinstance(j.get('input'), dict): return {'error': 'not { artifact, input } with an artifact this implementation judges'} return JUDGES[j['artifact']](j['input']) except Exception as e: # a crash is reported as such: the runner scores it BROKEN, never as a refusal return {'error': str(e)[:300]} def main(a): if '--capabilities' in a: print(json.dumps(CAPABILITIES)) return 0 if '--batch' in a: with open(a[a.index('--batch') + 1], encoding='utf-8') as f: for e in json.load(f): print(json.dumps(one(e['file']))) return 0 files = [x for x in a if not x.startswith('--')] if not files: print('usage: implementare-integritate.py --capabilities | | --batch ', file=sys.stderr) return 3 r = one(files[0]) print(json.dumps(r, indent=1)) return 3 if 'error' in r else {'VALID': 0, 'INVALID': 1}.get(r['verdict'], 2) if __name__ == '__main__': sys.exit(main(sys.argv[1:]))