aere-network/aere-quantum
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
82c5b261ac
aere-quantum/conformance/implementare-integritate.py

427 lines
22 KiB
Python
Raw Blame History

#!/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 <vector.json>
python3 implementare-integritate.py --batch <list.json>
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 | <vector.json> | --batch <list.json>', 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:]))
Reference in New Issue View Git Blame Copy Permalink