64 lines
2.6 KiB
Python
64 lines
2.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Offline schema references and frozen historical provenance checks."""
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
|
|
|
|
def digest(path):
|
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|
|
|
|
|
|
def verify_bundle(root):
|
|
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 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'))
|
|
return len(paths)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
try:
|
|
count = verify_bundle(Path(__file__).resolve().parent)
|
|
except (ValueError, OSError) as exc:
|
|
raise SystemExit(str(exc)) from exc
|
|
print(f'shared contract: {count} JSON files; offline references and frozen historical provenance valid')
|