"""NOIA release verifier. Stdlib only; also distributed as a standalone script. Hash profile: UTF-8, sorted Unicode code-point keys, compact JSON, no NaN, no Unicode normalization, one trailing LF. This is NOT RFC 8785/JCS. """ import argparse import hashlib import json import re from datetime import datetime from pathlib import Path PROFILE = 'noia-evidence-json-v1' VERSION = 'noia-release-evidence-v1' RELEASE_RE = r'[0-9]{4}\.[0-9]{2}\.[0-9]{2}\.[1-9][0-9]*' EVIDENCE = 'data/evidence/' def encoded(value): return (json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(',', ':'), allow_nan=False) + '\n').encode('utf8') def sha(data): return hashlib.sha256(data).hexdigest() def file_hash(path): result = hashlib.sha256() with Path(path).open('rb') as stream: for chunk in iter(lambda: stream.read(1024 * 1024), b''): result.update(chunk) return result.hexdigest() def _pairs(items): result = {} for key, value in items: if key in result: raise ValueError('Duplicate JSON key: ' + key) result[key] = value return result def loads(raw): return json.loads(raw, object_pairs_hook=_pairs, parse_constant=lambda x: (_ for _ in ()).throw(ValueError('Non-finite JSON'))) def read(path): return loads(Path(path).read_bytes()) def target(root, relative): if not isinstance(relative, str) or not relative or '\\' in relative or ':' in relative: raise ValueError('Unsafe evidence path') parts = relative.split('/') if any(part in ('', '.', '..') for part in parts): raise ValueError('Unsafe evidence path') root = Path(root).resolve() candidate = root for part in parts: candidate = candidate / part if candidate.is_symlink(): raise ValueError('Symlinks are not evidence files') if root not in candidate.resolve().parents: raise ValueError('Evidence path escapes root') return candidate def descriptor(root, relative): path = target(root, relative) return {'path': relative, 'bytes': path.stat().st_size, 'sha256': file_hash(path)} def content_payload(row, images): """Project published content without release URLs, timestamps or taxonomy. Image URLs are replaced by source blob commitments, retaining image order and the placement context. The source commitment is not a claim that its original bytes are included in this public dataset. """ replacements = {} for source, image in images.items(): token = 'archive-blob:sha256:' + image['source_sha256'] replacements[source] = token for kind in ('content', 'thumbnail'): replacements['/' + image[kind]['path']] = token # Only paths present in this record are replaced; longest first. def rewrite(value): if isinstance(value, str): # Public media paths are ASCII tokens without spaces or closing brackets. return re.sub(r'/?(?:media|branding)/[A-Za-z0-9_.-]+', lambda m: replacements.get(m.group(0), m.group(0)), value) if isinstance(value, list): return [rewrite(v) for v in value] if isinstance(value, dict): return {k: rewrite(v) for k, v in value.items()} return value figures = [] for figure in row.get('figures', []): source = figure.get('source_path') if source not in images: raise ValueError('Missing figure source commitment') if figure.get('display_derivative') != images[source] or figure.get('path') != '/' + images[source]['content']['path']: raise ValueError('Figure/image map mismatch') item = {k: v for k, v in figure.items() if k not in ('path', 'source_path', 'display_derivative')} item['source_sha256'] = images[source]['source_sha256'] figures.append(item) return {'hash_profile': PROFILE, 'kind': 'article-content', **rewrite({k: row.get(k) for k in ('title', 'markdown', 'blocks', 'sections', 'thumbnail', 'editorial')}), 'figures': rewrite(figures), 'audio': row.get('audio', []), 'omitted_media': row.get('omitted_media', [])} def revision_payload(row, integrity): return {'hash_profile': PROFILE, 'kind': 'public-article-revision', **{k: row.get(k) for k in ('id', 'work_id', 'edition_id', 'time', 'source_url', 'rights', 'package_sha256', 'legacy')}, **{k: integrity[k] for k in ('content_sha256', 'lineage', 'previous_revision_sha256', 'previous_revision_status')}} def article_integrity(row, images, lineage=(), previous=None): parent = next((p for p in lineage if p['relation'] == 'revision_parent'), None) result = {'hash_profile': PROFILE, 'content_sha256': sha(encoded(content_payload(row, images))), 'lineage': list(lineage), 'previous_revision_sha256': previous, 'previous_revision_status': 'linked' if previous else 'not_in_previous_release' if parent else 'no_parent'} result['revision_sha256'] = sha(encoded(revision_payload(row, result))) return result def article_rows(site): index = read(target(site, 'data/articles-index.json')) seen = set() for part in index['shards']: path = target(site, 'data/' + part['path']) if path.stat().st_size != part['bytes'] or file_hash(path) != part['sha256']: raise ValueError('Article shard hash mismatch') lines = path.read_bytes().splitlines() if len(lines) != part['count']: raise ValueError('Article shard count mismatch') for line in lines: row = loads(line) if row['id'] in seen: raise ValueError('Duplicate article ID') seen.add(row['id']) yield row, sha(line) if len(seen) != index['count']: raise ValueError('Article count mismatch') def _is_payload(relative): return relative.startswith(('data/', 'media/')) and not relative.startswith(EVIDENCE) def verify_release(site, expected_sha256=None): site = Path(site).resolve() manifest_path = target(site, EVIDENCE + 'release.json') digest = file_hash(manifest_path) if expected_sha256 and digest != expected_sha256: raise ValueError('Release differs from trusted manifest hash') manifest = read(manifest_path) if manifest.get('schema_version') != VERSION or manifest.get('hash_profile') != PROFILE: raise ValueError('Unknown evidence schema/profile') if not re.fullmatch(RELEASE_RE, manifest.get('release_id', '')): raise ValueError('Invalid release ID') if datetime.fromisoformat(manifest['issued_at'].replace('Z', '+00:00')).tzinfo is None: raise ValueError('Release issuance needs an offset') if manifest.get('license_ref') is not None or manifest.get('license_status') != 'not_specified': raise ValueError('Unexpected evidence license policy') files = manifest['files'] paths = [f['path'] for f in files] if len(set(paths)) != len(paths): raise ValueError('Duplicate manifest path') for entry in files: if descriptor(site, entry['path']) != entry: raise ValueError('Release file changed: ' + entry['path']) actual = {p.relative_to(site).as_posix() for folder in ('data', 'media') for p in (site / folder).rglob('*') if p.is_file() and _is_payload(p.relative_to(site).as_posix())} if actual != {p for p in paths if _is_payload(p)}: raise ValueError('Release payload file set differs') required = {EVIDENCE + name for name in ('articles.jsonl', 'research.jsonl', 'verify_release.py', 'README.md')} if not required <= set(paths): raise ValueError('Evidence dependencies missing') current = manifest chain = [] previous_rows = [] while current.get('previous_release'): previous = current['previous_release'] previous_id = previous['release_id'] if previous_id in chain or not re.fullmatch(RELEASE_RE, previous_id): raise ValueError('Cyclic/invalid release chain') if tuple(map(int, previous_id.split('.'))) >= tuple(map(int, current['release_id'].split('.'))): raise ValueError('Release sequence goes backwards') relative = EVIDENCE + 'history/' + previous_id + '.json' if relative not in paths: raise ValueError('Uncommitted release history') old = target(site, relative) if file_hash(old) != previous['manifest_sha256']: raise ValueError('Previous manifest hash mismatch') current = read(old) if current['release_id'] != previous_id or current.get('schema_version') != VERSION: raise ValueError('Previous manifest identity differs') chain.append(previous_id) ledger_path = EVIDENCE + 'history/' + previous_id + '.articles.jsonl' ledger_descriptor = next((f for f in current['files'] if f['path'] == EVIDENCE + 'articles.jsonl'), None) if ledger_path not in paths or not ledger_descriptor: raise ValueError('Previous article evidence missing') ledger_file = target(site, ledger_path) if file_hash(ledger_file) != ledger_descriptor['sha256'] or ledger_file.stat().st_size != ledger_descriptor['bytes']: raise ValueError('Previous article evidence differs from its manifest') previous_rows.extend(loads(line) for line in ledger_file.read_bytes().splitlines()) images = read(target(site, 'data/image-map.json')) for value in images.values(): for kind in ('content', 'thumbnail'): image = value[kind] if image['path'] not in paths or descriptor(site, image['path']) != {k: image[k] for k in ('path', 'bytes', 'sha256')}: raise ValueError('Image commitment mismatch') ledger = [loads(line) for line in target(site, EVIDENCE + 'articles.jsonl').read_bytes().splitlines()] expected = [] for row, record_hash in article_rows(site): integrity = row.get('integrity', {}) computed = article_integrity(row, images, integrity.get('lineage', ()), integrity.get('previous_revision_sha256')) if integrity != computed: raise ValueError('Article content/revision hash mismatch: ' + row['id']) if row['dataset_version'] != manifest['release_id']: raise ValueError('Record release version mismatch') if integrity['previous_revision_sha256']: parent = next((p for p in integrity['lineage'] if p['relation'] == 'revision_parent'), None) if not parent or not any(old['id'] == parent['revision_id'] and old['revision_sha256'] == integrity['previous_revision_sha256'] and old['work_id'] == row['work_id'] and old['edition_id'] == row['edition_id'] for old in previous_rows): raise ValueError('Article parent is not committed in release history') expected.append({'id': row['id'], 'work_id': row['work_id'], 'edition_id': row['edition_id'], 'record_sha256': record_hash, **computed}) if sorted(ledger, key=lambda r: r['id']) != sorted(expected, key=lambda r: r['id']): raise ValueError('Article evidence ledger mismatch') research_path = site / 'data/research/cards.json' cards = read(research_path) if research_path.exists() else [] research = [{'id': c['id'], 'revision_id': c['revision_id'], 'code': c['code'], 'hash_profile': PROFILE, 'record_sha256': sha(encoded(c))} for c in cards] stored = [loads(line) for line in target(site, EVIDENCE + 'research.jsonl').read_bytes().splitlines()] if stored != sorted(research, key=lambda c: c['id']): raise ValueError('Research evidence ledger mismatch') if manifest['article_count'] != len(expected) or manifest['research_count'] != len(cards): raise ValueError('Evidence record totals differ') return {'status': 'PASS', 'release_id': manifest['release_id'], 'manifest_sha256': digest, 'files_checked': len(paths), 'articles_checked': len(expected), 'research_checked': len(cards), 'previous_releases_checked': len(chain), 'trusted_hash_supplied': bool(expected_sha256), 'scope': 'byte_integrity_and_history; external_timestamp_and_signature_require_separate_verification'} if __name__ == '__main__': parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('site', type=Path) parser.add_argument('--expected-sha256', help='Manifest digest obtained through an independent trusted channel') args = parser.parse_args() try: print(json.dumps(verify_release(args.site, args.expected_sha256), ensure_ascii=False, indent=2)) except (ValueError, OSError, KeyError) as exc: parser.exit(2, 'Verification failed: ' + str(exc) + '\n')