#!/usr/bin/env python3 """National AI Feed verifier (Python). Verification rules, version 1.2. Checks a stream of feed pages with the public key file alone. It needs the 'cryptography' package. Usage: python3 verify_feed.py [--keys public-key.json] [--prev-seq N --prev-hash HEX] [--filtered] [--json] page1.json [page2.json ...] --filtered the pages came from a filtered read (authority_record_id, provider_id or government). The chain cannot be followed entry to entry and completeness cannot be shown; see VERIFICATION.md. python3 verify_feed.py --vectors DIR run every test vector in DIR and report agreement A page is a /v1/feed response ({entries, next_after, has_more, feed_head}). A single entry or a list of entries is also accepted (then there is no signed head). Exit status: 0 nothing wrong found, 1 problems found, 2 input could not be read. """ import sys, os, json, re, hashlib, glob SPEC = '1.2' NEW_FIELDS = [('authority_record_id', 'i'), ('provider_id', 'i'), ('provider_name', 's'), ('government_official_id_type', 's'), ('government_official_id', 's'), ('revision', 'i'), ('supersedes_seq', 'i'), ('prev_seq', 'i'), ('prev_hash', 's'), ('status_scope', 's'), ('status_event', 's'), ('reason_class', 's'), ('reason_code', 's'), ('revoked_from_seq', 'i'), ('revoked_through_seq', 'i'), ('references_seq', 'i')] INT_FIELDS = {k for k, t in NEW_FIELDS if t == 'i'} STR_FIELDS = {k for k, t in NEW_FIELDS if t == 's'} SCOPE_EVENTS = {'provider': {'suspended', 'reinstated', 'revoked', 'revocation_lifted'}, 'authority_record': {'suspended', 'reinstated', 'ended', 'revoked', 'revocation_lifted'}, 'publication': {'revoked', 'revocation_lifted'}} ADMIN = {'agreement_lapsed', 'provider_request', 'investigation', 'technical_problem', 'other'} COMPROMISE = {'unauthorized_publishing', 'fraud', 'key_compromise'} HEX64 = re.compile(r'[0-9a-f]{64}\Z') HEX128 = re.compile(r'[0-9a-fA-F]{128}\Z') UTC = re.compile(r'[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}Z\Z') REVOKE_REASONS = {'key_compromise', 'suspected_compromise'} LONE_SURROGATE = re.compile('[\ud800-\udfff]') class Unreadable(Exception): pass # ---------------------------------------------------------------- canonical JSON (the exact text that is hashed) def enc_str(s): out = ['"'] for ch in s: o = ord(ch) if ch == '"': out.append('\\"') elif ch == '\\': out.append('\\\\') elif ch == '\b': out.append('\\b') elif ch == '\f': out.append('\\f') elif ch == '\n': out.append('\\n') elif ch == '\r': out.append('\\r') elif ch == '\t': out.append('\\t') elif o < 0x20: out.append('\\u%04x' % o) elif o == 0x2028: out.append('\\u2028') elif o == 0x2029: out.append('\\u2029') else: out.append(ch) out.append('"') return ''.join(out) def enc(v): if v is None: return 'null' if isinstance(v, bool): return 'true' if v else 'false' if isinstance(v, int): return str(v) if isinstance(v, str): return enc_str(v) raise Unreadable('unsupported value in canonical form') def blank_null(v): return None if v == '' else v def canonical(e): a = e.get('authority') if isinstance(e.get('authority'), dict) else {} f = [('seq', e['seq']), ('type', e['type']), ('external_id', e.get('external_id')), ('authority_name', a.get('name')), ('authority_type', a.get('type')), ('state', a.get('state')), ('county', a.get('county')), ('city', a.get('city')), ('agency', e.get('agency')), ('title', e.get('title')), ('body', e.get('body')), ('source_url', e.get('source_url')), ('published_at', e.get('published_at')), ('received_at', e.get('received_at'))] parts = ['%s:%s' % (enc_str(k), enc(blank_null(v))) for k, v in f] att = e.get('attachments') if isinstance(att, list) and att: items = [] for x in att: items.append('{%s}' % ','.join('%s:%s' % (enc_str(k), enc(x.get(k))) for k in ('id', 'filename', 'content_type', 'size', 'sha256'))) parts.append('%s:[%s]' % (enc_str('attachments'), ','.join(items))) for k, t in NEW_FIELDS: v = e.get(k) if v is None or v == '': continue parts.append('%s:%s' % (enc_str(k), enc(v))) return '{' + ','.join(parts) + '}' # ---------------------------------------------------------------- crypto HEX64_ANY = re.compile(r'[0-9a-fA-F]{64}\Z') SMALL_ORDER = frozenset(( '0000000000000000000000000000000000000000000000000000000000000000', '0000000000000000000000000000000000000000000000000000000000000080', '0100000000000000000000000000000000000000000000000000000000000000', '0100000000000000000000000000000000000000000000000000000000000080', '26e8958fc2b227b045c3f489f2ef98f0d5dfac05d3c63339b13802886d53fc05', '26e8958fc2b227b045c3f489f2ef98f0d5dfac05d3c63339b13802886d53fc85', 'c7176a703d4dd84fba3c0b760d10670f2a2053fa2c39ccc64ec7fd7792ac037a', 'c7176a703d4dd84fba3c0b760d10670f2a2053fa2c39ccc64ec7fd7792ac03fa', 'ecffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff7f', 'ecffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff', 'edffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff7f', 'edffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff', 'eeffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff7f', 'eeffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff', )) # rules 1.2 section 1.4: points of small order (all 8, both sign bits, and the non-canonical forms). A key like this lets a forged signature pass in OpenSSL-based libraries. def load_keys(doc): """returns (keys, raw): the usable public keys by key_id, and the key-file entry each came from. A key-file entry is usable only if it is an object whose key_id is non-empty text and whose public_key is 64 hex characters; anything else is skipped. A key file whose 'keys' is not a list has no usable keys.""" from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey keys, raw = {}, {} lst = doc.get('keys') if isinstance(doc, dict) else None for k in lst if isinstance(lst, list) else []: if not isinstance(k, dict) or not isinstance(k.get('key_id'), str) or k['key_id'] == '' or not isinstance(k.get('public_key'), str) or not HEX64_ANY.match(k['public_key']) or k['public_key'].lower() in SMALL_ORDER: continue try: keys[k['key_id']] = Ed25519PublicKey.from_public_bytes(bytes.fromhex(k['public_key'])) raw[k['key_id']] = k except Exception: pass return keys, raw def notice_text(kid, k): """the exact text a revocation notice signs (VERIFICATION.md 1.3)""" lth = k.get('last_trusted_hash') return '{"type":"key_revocation","key_id":%s,"public_key":%s,"revoked_at":%s,"compromised_since":%s,"last_trusted_seq":%d,"last_trusted_hash":%s,"reason":%s,"signed_by":%s}' % ( enc_str(kid), enc_str(k['public_key']), enc_str(k['revoked_at']), enc_str(k['compromised_since']), k['last_trusted_seq'], 'null' if lth is None else enc_str(lth), enc_str(k['reason']), enc_str(k['notice_signed_by'])) def load_revocations(keys, raw): """for every key the file lists as revoked: whether its notice is valid and what it says. A notice that is not valid makes every entry of that key untrusted.""" out = {} for kid, k in raw.items(): if not isinstance(kid, str) or k.get('status') != 'revoked': continue info = {'valid': False, 'why': '', 'lts': 0, 'lth': None, 'since': None} out[kid] = info ra, cs, lts, lth, why_, by, sig = (k.get(x) for x in ('revoked_at', 'compromised_since', 'last_trusted_seq', 'last_trusted_hash', 'reason', 'notice_signed_by', 'notice_signature')) if not (isinstance(ra, str) and UTC.match(ra) and isinstance(cs, str) and UTC.match(cs)): info['why'] = 'revoked_at or compromised_since is missing or not a UTC time'; continue if cs > ra: info['why'] = 'compromised_since is after revoked_at'; continue if not is_int(lts) or lts < 0: info['why'] = 'last_trusted_seq is not a whole number of zero or more'; continue if lts == 0 and lth is not None: info['why'] = 'last_trusted_hash must be null when nothing is trusted'; continue if lts >= 1 and not (isinstance(lth, str) and HEX64.match(lth)): info['why'] = 'last_trusted_hash is not a 64-character lower-case hash'; continue if not isinstance(why_, str) or why_ not in REVOKE_REASONS: info['why'] = 'reason is not key_compromise or suspected_compromise'; continue if not isinstance(by, str) or by == kid or by not in keys: info['why'] = 'notice_signed_by is missing, unknown, or the revoked key itself'; continue if not isinstance(sig, str) or not HEX128.match(sig): info['why'] = 'notice_signature is missing'; continue h = hashlib.sha256(notice_text(kid, k).encode('utf-8')).hexdigest() if not sig_ok(keys[by], sig, h): info['why'] = 'notice_signature does not verify with ' + by; continue info.update(valid=True, lts=lts, lth=lth, since=cs) return out def sig_ok(key, sig_hex, text): try: if not HEX128.match(sig_hex): return False key.verify(bytes.fromhex(sig_hex), text.encode('ascii')) return True except Exception: return False # ---------------------------------------------------------------- shape def is_int(v): return isinstance(v, int) and not isinstance(v, bool) def walk_strings(o): if isinstance(o, str): if LONE_SURROGATE.search(o): raise Unreadable('a string holds a lone surrogate (not valid text)') elif isinstance(o, dict): for k, v in o.items(): walk_strings(k); walk_strings(v) elif isinstance(o, list): for v in o: walk_strings(v) def entry_shape_problem(e): if not isinstance(e, dict): return 'entry is not an object' if not is_int(e.get('seq')) or e['seq'] < 1: return 'seq is not a positive integer' if e.get('type') not in ('record', 'removal', 'status'): return 'type is not record, removal or status' for k in ('record_hash', 'signature', 'public_key_id', 'received_at'): if not isinstance(e.get(k), str): return k + ' is missing' if e.get('external_id') is not None and not isinstance(e.get('external_id'), str): return 'external_id is not text' a = e.get('authority') if not isinstance(a, dict): return 'authority is missing' for k in ('name', 'type', 'state', 'county', 'city'): if a.get(k) is not None and not isinstance(a.get(k), str): return 'authority.' + k + ' is not text' for k in ('agency', 'title', 'body', 'source_url', 'published_at'): if e.get(k) is not None and not isinstance(e.get(k), str): return k + ' is not text' for k in INT_FIELDS: if e.get(k) is not None and not is_int(e.get(k)): return k + ' is not an integer' for k in STR_FIELDS: if e.get(k) is not None and not isinstance(e.get(k), str): return k + ' is not text' att = e.get('attachments') if att is not None: if not isinstance(att, list): return 'attachments is not a list' for x in att: if not isinstance(x, dict): return 'an attachment is not an object' for k in ('id', 'filename', 'content_type', 'sha256'): if not isinstance(x.get(k), str): return 'attachment ' + k + ' is missing' if not is_int(x.get('size')): return 'attachment size is not an integer' return None def status_problem(e, first_seq, by_seq): s, ev = e.get('status_scope'), e.get('status_event') if s not in SCOPE_EVENTS or ev not in SCOPE_EVENTS[s]: return 'scope and event do not go together' if e.get('provider_id') is None or not e.get('provider_name'): return 'no provider named' if e.get('title') is not None or e.get('body') is not None or e.get('revision') is not None or e.get('supersedes_seq') is not None or e.get('attachments'): return 'carries publication fields' if s == 'provider' and (e.get('authority_record_id') is not None or e.get('government_official_id') not in (None, '')): return 'provider-level entry names a record or government' if s != 'provider' and (e.get('authority_record_id') is None or not e.get('government_official_id')): return 'record-level entry names no record or government' if s == 'publication' and not e.get('external_id'): return 'publication-level entry has no external_id' if s != 'publication' and e.get('external_id'): return 'unexpected external_id' if ev == 'revoked': f, t = e.get('revoked_from_seq'), e.get('revoked_through_seq') if e.get('reason_class') != 'compromise' or e.get('reason_code') not in COMPROMISE: return 'revocation reason is not a compromise reason' if f is None or t is None or not (1 <= f <= t < e['seq']): return 'revoked range is not valid' if e.get('references_seq') is not None: return 'revocation carries references_seq' elif ev == 'revocation_lifted': if e.get('reason_class') != 'administrative' or e.get('reason_code') != 'error_corrected': return 'lift reason is not error_corrected' if e.get('revoked_from_seq') is not None or e.get('revoked_through_seq') is not None: return 'lift carries a range' r = e.get('references_seq') if r is None or not (1 <= r < e['seq']): return 'lift does not name an earlier entry' else: if e.get('reason_class') != 'administrative' or e.get('revoked_from_seq') is not None or e.get('revoked_through_seq') is not None or e.get('references_seq') is not None: return 'administrative event has revocation fields' if ev == 'suspended' and e.get('reason_code') not in ADMIN: return 'suspension reason not allowed' if ev == 'ended' and e.get('reason_code') != 'provider_request': return 'end reason not allowed' if ev == 'reinstated' and e.get('reason_code') is not None: return 'reinstatement carries a reason' return None def covers(r, x): """Does revocation entry r cover entry x (not a status entry)?""" if x['type'] == 'status': return False if not (r['revoked_from_seq'] <= x['seq'] <= r['revoked_through_seq']): return False s = r['status_scope'] if s == 'provider': return x.get('provider_id') == r['provider_id'] if s == 'authority_record': return x.get('authority_record_id') == r['authority_record_id'] return x.get('authority_record_id') == r['authority_record_id'] and x.get('external_id') == r['external_id'] # ---------------------------------------------------------------- the verification def verify_stream(pages, keys_doc, anchor=None, filtered=False): """filtered=True: the pages came from a filtered read (authority_record_id, provider_id or government). See VERIFICATION.md, 'Filtered reads'.""" if filtered and anchor: raise Unreadable('an anchor cannot be used with a filtered read (it skips entries, so no entry follows the anchor)') keys, raw_keys = load_keys(keys_doc) revk = load_revocations(keys, raw_keys) P, N = [], [] def prob(code, seq=None, detail=''): P.append({'code': code, 'seq': seq, 'detail': detail}) def note(code, seq=None): N.append({'code': code, 'seq': seq}) if filtered: note('filtered_read') for kid, r in revk.items(): if not r['valid']: prob('revocation_notice_invalid', None, 'key %s: %s' % (kid, r['why'])) by_rev = {} # revoked key id -> seqs of entries it signed correctly entries = [] heads = [] # (page index, head, last seq of that page, has_more) last_page = None for pi, pg in enumerate(pages): if isinstance(pg, dict) and 'entries' in pg: page = pg elif isinstance(pg, dict) and 'seq' in pg: page = {'entries': [pg]} elif isinstance(pg, list): page = {'entries': pg} else: raise Unreadable('page %d is not a feed page, an entry or a list of entries' % (pi + 1)) if not isinstance(page['entries'], list): raise Unreadable('entries is not a list') good = [] for e in page['entries']: sp = entry_shape_problem(e) if sp: prob('entry_malformed', e.get('seq') if isinstance(e, dict) and is_int(e.get('seq')) and e['seq'] >= 1 else None, sp) else: good.append(e) raw_last = page['entries'][-1] if page['entries'] else None if isinstance(raw_last, dict) and is_int(raw_last.get('seq')) and 'next_after' in page and not (is_int(page['next_after']) and page['next_after'] == raw_last['seq']): prob('page_cursor_mismatch', raw_last['seq'], 'next_after is not the last entry') entries.extend(good) last_page = page if page.get('feed_head') is not None: heads.append((pi, page['feed_head'], good[-1]['seq'] if good else None)) by_seq = {} first_seq = entries[0]['seq'] if entries else None prev = None pub = {} # (authority_record_id, external_id) -> (seq, revision) revokes = [] # status entries that revoke lifted = {} # references_seq -> lifting seq for idx, e in enumerate(entries): seq = e['seq'] # signature hash_ok = False text = canonical(e) h = hashlib.sha256(text.encode('utf-8')).hexdigest() if e.get('signature_algorithm') not in (None, 'Ed25519'): prob('algorithm_unsupported', seq) elif h != e['record_hash'] or not HEX64.match(e['record_hash']): prob('hash_mismatch', seq) else: hash_ok = True k = keys.get(e['public_key_id']) if k is None: prob('key_unknown', seq) elif not sig_ok(k, e['signature'], h): prob('signature_invalid', seq) elif e['public_key_id'] in revk: by_rev.setdefault(e['public_key_id'], []).append(seq) # order if prev is not None and seq <= prev['seq']: prob('seq_not_increasing', seq) # chain chained = e.get('prev_seq') is not None if filtered: # a filtered read skips entries, so links cannot be followed one after another; each link is checked where it can be if chained: ps = e['prev_seq'] if not is_int(ps) or ps >= seq: prob('chain_break', seq, 'prev_seq is not an earlier entry') elif ps in by_seq and e.get('prev_hash') != by_seq[ps]['record_hash']: prob('chain_break', seq, 'prev_hash differs from the hash of entry %d in this stream' % ps) elif prev is not None and prev.get('prev_seq') is not None: prob('chain_missing', seq) else: note('unchained_entry', seq) elif prev is None: if anchor and anchor.get('prev_seq') is not None: if e.get('prev_seq') != anchor['prev_seq'] or e.get('prev_hash') != anchor.get('prev_hash'): prob('chain_anchor_mismatch', seq) elif chained: note('start_not_anchored', seq) else: note('unchained_entry', seq) else: if chained: if e.get('prev_seq') != prev['seq'] or e.get('prev_hash') != prev['record_hash']: prob('chain_break', seq) elif prev.get('prev_seq') is not None: prob('chain_missing', seq) else: note('unchained_entry', seq) # publications if e['type'] in ('record', 'removal') and e.get('authority_record_id') is not None: rev, sup = e.get('revision'), e.get('supersedes_seq') if rev is None or e.get('provider_id') is None or not e.get('government_official_id'): prob('entry_malformed', seq, 'publication entry lacks revision, provider or government') else: key = (e['authority_record_id'], e.get('external_id')) if key in pub: ls, lr = pub[key] if rev != lr + 1: prob('revision_gap', seq) if sup != ls: prob('supersedes_mismatch', seq) else: if rev == 1: if sup is not None: prob('supersedes_mismatch', seq) else: note('revision_start_not_seen', seq) pub[key] = (seq, rev if rev is not None else 0) # status if e['type'] == 'status': why = status_problem(e, first_seq, by_seq) if not why and not filtered and e['status_event'] == 'revoked' and first_seq is not None and first_seq <= e['revoked_from_seq']: if not any(e['revoked_from_seq'] <= x['seq'] <= e['revoked_through_seq'] and covers(e, x) for x in entries[:idx]): why = 'range holds no entry of its scope' if not why and e['status_event'] == 'revocation_lifted': r = by_seq.get(e['references_seq']) if r is not None: if r['type'] != 'status' or r.get('status_event') != 'revoked' or r.get('status_scope') != e['status_scope'] or r.get('provider_id') != e['provider_id'] \ or r.get('authority_record_id') != e.get('authority_record_id') or r.get('external_id') != e.get('external_id'): why = 'does not lift a revocation of the same scope' elif e['references_seq'] in lifted: prob('lift_duplicate', seq) elif not filtered and first_seq is not None and e['references_seq'] >= first_seq: why = 'lifts an entry that is not in the feed' else: note('lift_target_not_seen', seq) if why: prob('status_malformed', seq, why) elif e['status_event'] == 'revoked': revokes.append(e) elif e['status_event'] == 'revocation_lifted': lifted.setdefault(e['references_seq'], seq) by_seq[seq] = e prev = e # entries signed by a revoked key: trusted only up to the last trusted entry named in the signed notice, and only if the stream reaches that entry with that hash byseq_hash = {e['seq']: e['record_hash'] for e in entries} for kid, seqs in by_rev.items(): r = revk[kid] if not r['valid']: for sq in seqs: prob('key_revoked', sq, 'key %s is listed as revoked with a notice that is not valid, so nothing it signed can be trusted' % kid) continue for sq in seqs: if sq > r['lts']: prob('key_revoked', sq, 'signed by revoked key %s after its last trusted entry (seq %d)' % (kid, r['lts'])) before = [sq for sq in seqs if sq <= r['lts']] if before: if r['lts'] in byseq_hash: if byseq_hash[r['lts']] == r['lth']: note('signed_by_revoked_key_trusted_prefix', min(before)) else: for sq in before: prob('key_revoked', sq, 'the stream holds entry %d with a different hash than the revocation notice of key %s names' % (r['lts'], kid)) else: note('revoked_key_prefix_unconfirmed', min(before)) # heads for pi, hd, lastseq in heads: if not isinstance(hd, dict) or not is_int(hd.get('seq')) or not isinstance(hd.get('record_hash'), str) or not isinstance(hd.get('signed_at'), str) \ or not isinstance(hd.get('head_hash'), str) or not isinstance(hd.get('signature'), str) or not isinstance(hd.get('public_key_id'), str): prob('head_malformed', None); continue text = '{"type":"feed_head","seq":%d,"record_hash":%s,"signed_at":%s}' % (hd['seq'], enc_str(hd['record_hash']), enc_str(hd['signed_at'])) hh = hashlib.sha256(text.encode('utf-8')).hexdigest() if hh != hd['head_hash']: prob('head_hash_mismatch', hd['seq']) else: k = keys.get(hd['public_key_id']) if k is None: prob('head_key_unknown', hd['seq']) elif not sig_ok(k, hd['signature'], hh): prob('head_signature_invalid', hd['seq']) elif hd['public_key_id'] in revk: r = revk[hd['public_key_id']] if not r['valid'] or hd['seq'] > r['lts'] or not UTC.match(hd['signed_at']) or hd['signed_at'] >= r['since']: prob('head_key_revoked', hd['seq']) if hd['seq'] in byseq_hash and byseq_hash[hd['seq']] != hd['record_hash']: prob('head_entry_mismatch', hd['seq']) if lastseq is not None and hd['seq'] < lastseq: prob('head_behind', hd['seq']) if not heads: note('no_head') else: pi, hd, lastseq = heads[-1] if pi == len(pages) - 1 and isinstance(hd, dict) and is_int(hd.get('seq')): newest = max(e['seq'] for e in entries) if entries else None if last_page.get('has_more') is True: note('stream_incomplete') elif not filtered and newest is not None and hd['seq'] > newest: prob('tail_truncated', hd['seq']) # derived: which entries stand revoked and by which status entry (an unlifted revocation that covers them; the earliest) revoked = {} for x in entries: if x['type'] == 'status': continue for r in revokes: if r['seq'] in lifted: continue if covers(r, x): revoked[str(x['seq'])] = r['seq']; break P.sort(key=lambda p: (p['seq'] if p['seq'] is not None else -1, p['code'])) N.sort(key=lambda p: (p['seq'] if p['seq'] is not None else -1, p['code'])) return {'verifier': 'python', 'spec': SPEC, 'entries': len(entries), 'ok': not P, 'problems': P, 'notes': N, 'derived': {'revoked': revoked}} def pairs(lst): return [[x['code'], x['seq']] for x in lst] # ---------------------------------------------------------------- vectors def run_vector(v): try: pages = [parse_text(t, 'vector') for t in v['raw_pages']] if 'raw_pages' in v else v['pages'] walk_strings(pages) rep = verify_stream(pages, v['keys'], v.get('anchor'), bool(v.get('filtered'))) except Unreadable: rep = {'unreadable': True} exp = v['expect'] if exp.get('unreadable'): return rep.get('unreadable') is True, rep if rep.get('unreadable'): return False, rep ok = rep['ok'] == exp['ok'] and pairs(rep['problems']) == exp['problems'] and pairs(rep['notes']) == exp['notes'] and rep['derived'] == exp['derived'] return ok, rep def run_vectors(d): files = sorted(glob.glob(os.path.join(d, '*.json'))) files = [f for f in files if os.path.basename(f) != 'manifest.json'] bad = 0 for f in files: v = json.load(open(f, encoding='utf-8')) ok, rep = run_vector(v) print(('PASS ' if ok else 'FAIL ') + os.path.basename(f)) if not ok: bad += 1 print(' got:', json.dumps({k: rep.get(k) for k in ('ok', 'problems', 'notes', 'derived', 'unreadable')})[:600]) print('%d vectors, %d passed, %d failed' % (len(files), len(files) - bad, bad)) return 1 if bad else 0 MAX_DEPTH = 64 MAX_SAFE = 2 ** 53 - 1 _LEX = re.compile(r'"[^"\\]*(?:\\.[^"\\]*)*"|[\[\]{}]|-?[0-9][0-9.eE+\-]*|-|NaN|Infinity', re.S) _INTTOK = re.compile(r'-?(?:0|[1-9][0-9]*)') def lex_check(text, path): """rules 1.2 section 1.4: before parsing, refuse nesting deeper than 64 levels and any number that is not a whole number within +-(2^53-1) (no fractions, exponents, NaN or Infinity). The three verifiers apply this identically, so they read the same numbers.""" depth = 0 for m in _LEX.finditer(text): t = m.group() c = t[0] if c == '"': continue if c in '[{': depth += 1 if depth > MAX_DEPTH: raise Unreadable('%s: nested deeper than %d levels' % (path, MAX_DEPTH)) elif c in ']}': depth -= 1 elif not _INTTOK.fullmatch(t) or abs(int(t)) > MAX_SAFE: raise Unreadable('%s: a number is not a whole number within +-(2^53-1)' % path) def parse_text(text, path): try: lex_check(text, path) o = json.loads(text) except Unreadable: raise except Exception as ex: raise Unreadable('%s: %s' % (path, ex)) walk_strings(o) return o def read_json(path): try: with open(path, encoding='utf-8') as fh: text = fh.read() except Exception as ex: raise Unreadable('%s: %s' % (path, ex)) return parse_text(text, path) def main(argv): keys_path, anchor, as_json, files, vec, filtered = 'public-key.json', {}, False, [], None, False i = 0 while i < len(argv): a = argv[i] if a == '--keys': keys_path = argv[i + 1]; i += 2 elif a == '--prev-seq': anchor['prev_seq'] = int(argv[i + 1]); i += 2 elif a == '--prev-hash': anchor['prev_hash'] = argv[i + 1]; i += 2 elif a == '--json': as_json = True; i += 1 elif a == '--filtered': filtered = True; i += 1 elif a == '--vectors': vec = argv[i + 1]; i += 2 else: files.append(a); i += 1 if vec: return run_vectors(vec) if '--canon-hashes' in argv: # test aid: print the hash of the canonical form of each entry in a list file for e in read_json(argv[argv.index('--canon-hashes') + 1]): print(hashlib.sha256(canonical(e).encode('utf-8')).hexdigest()) return 0 if not files: sys.stderr.write(__doc__); return 2 try: keys_doc = read_json(keys_path) pages = [read_json(f) for f in files] rep = verify_stream(pages, keys_doc, anchor or None, filtered) except Unreadable as ex: sys.stderr.write('Could not read input: %s\n' % ex); return 2 if as_json: print(json.dumps(rep, indent=1)) else: print('%d entries; %s' % (rep['entries'], 'no problems found' if rep['ok'] else '%d problem(s)' % len(rep['problems']))) for p in rep['problems']: print(' PROBLEM %-24s seq %s %s' % (p['code'], p['seq'], p['detail'])) for n in rep['notes']: if n['code'] != 'unchained_entry': print(' note %-24s seq %s' % (n['code'], n['seq'])) if rep['derived']['revoked']: print(' revoked entries (seq: revoked by status seq):', rep['derived']['revoked']) return 0 if rep['ok'] else 1 if __name__ == '__main__': sys.exit(main(sys.argv[1:]))