77 lines
3.4 KiB
Python
77 lines
3.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Offline reference/provenance checks; --write updates current document/bundle hashes."""
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
|
|
|
|
def digest(path):
|
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|
|
|
|
|
|
def verify_bundle(root, write=False):
|
|
shared = root / 'schema'
|
|
manifest_path = root / 'manifest.json'
|
|
manifest = json.loads(manifest_path.read_text(encoding='utf-8'))
|
|
for relative, expected in manifest['sources'].items():
|
|
actual = digest(root / relative)
|
|
if write and relative == 'schema/saas-dispatcher.md':
|
|
manifest['sources'][relative] = actual
|
|
elif actual != expected:
|
|
raise ValueError(f'source hash mismatch: {relative}')
|
|
|
|
schemas = {p.name: json.loads(p.read_text(encoding='utf-8')) for p in shared.glob('*.schema.json')}
|
|
def check_refs(value, name):
|
|
if isinstance(value, list):
|
|
for item in value:
|
|
check_refs(item, name)
|
|
elif isinstance(value, dict):
|
|
if '$ref' in value:
|
|
ref = value['$ref']
|
|
filename, _, pointer = ref.partition('#')
|
|
target_name = filename.removeprefix('./') if filename else name
|
|
if target_name not in schemas:
|
|
raise ValueError(f'missing schema reference: {name}: {ref}')
|
|
target = schemas[target_name]
|
|
if pointer:
|
|
if not pointer.startswith('/'):
|
|
raise ValueError(f'invalid schema pointer: {name}: {ref}')
|
|
try:
|
|
for part in pointer[1:].split('/'):
|
|
key = part.replace('~1', '/').replace('~0', '~')
|
|
target = target[int(key)] if isinstance(target, list) else target[key]
|
|
except (KeyError, ValueError, IndexError, TypeError) as exc:
|
|
raise ValueError(f'invalid schema pointer: {name}: {ref}') from exc
|
|
for item in value.values():
|
|
check_refs(item, name)
|
|
for name, schema in schemas.items():
|
|
check_refs(schema, name)
|
|
|
|
paths = sorted((p for p in shared.rglob('*.json') if
|
|
p.parent == shared or p.relative_to(shared).parts[0] == 'examples'),
|
|
key=lambda p: p.relative_to(shared).as_posix())
|
|
if len(paths) < 10:
|
|
raise ValueError('missing contract examples or schemas')
|
|
for path in paths:
|
|
json.loads(path.read_text(encoding='utf-8'))
|
|
listing = ''.join(f'{p.relative_to(shared).as_posix()} {digest(p)}\n' for p in paths)
|
|
actual = hashlib.sha256(listing.encode('utf-8')).hexdigest()
|
|
if write:
|
|
manifest['bundle_sha256'] = actual
|
|
manifest_path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')
|
|
elif actual != manifest['bundle_sha256']:
|
|
raise ValueError(f'contract bundle hash mismatch: {actual} != {manifest["bundle_sha256"]}')
|
|
return len(paths)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--write', action='store_true')
|
|
args = parser.parse_args()
|
|
try:
|
|
count = verify_bundle(Path(__file__).resolve().parent, args.write)
|
|
except (ValueError, OSError) as exc:
|
|
raise SystemExit(str(exc)) from exc
|
|
print(f'shared contract: {count} JSON files; source hashes, bundle hash and offline references valid')
|