224 lines
10 KiB
Python
224 lines
10 KiB
Python
#!/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)
|
|
|
|
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. It never opens a connection: --online is not claimed.
|
|
"""
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import sys
|
|
|
|
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')
|
|
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)
|
|
|
|
|
|
JUDGES = {'envelope': judge_envelope, 'dataset-record': judge_record}
|
|
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'], '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:]))
|