coulomb-social/scripts/verify-bubble-body-resources.py

#!/usr/bin/env python3
"""Verify retained body references through the product renderer, without network or import."""

import argparse
import hashlib
import importlib.util
import json
import os
import sys
from collections import Counter
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
os.environ['DJANGO_SETTINGS_MODULE'] = 'coulomb_social.settings.test'
import django  # noqa: E402

django.setup()
from coulomb_social.apps.spaces.bubble_bundle import prepare_bundle  # noqa: E402
from coulomb_social.apps.spaces.rendering_audit import audit_body  # noqa: E402

# Same bounded, owned-file and no-symlink reader used for the independent
# Visual-only preservation proof. It performs no initialization or network I/O.
SPEC = importlib.util.spec_from_file_location(
    'visual_preservation', ROOT / 'scripts/verify-bubble-visual-preservation.py')
reader = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(reader)


def sha(raw):
    return hashlib.sha256(raw).hexdigest()


def verify(original, prepared):
    raw_report = reader.read(original, 'reconciliation.json')
    before = json.loads(raw_report)
    visual_raw = reader.read(prepared, 'reconciliation.json')
    visual = json.loads(visual_raw)
    if (before['schema'] != 'coulomb.bubble-review-bundle.v1'
            or before['status'] != 'completed_offline_preparation'
            or visual['original_report_sha256'] != sha(raw_report)):
        raise ValueError('source report mismatch')
    inputs = {}
    for name in ['spaces', 'articles', 'retained_spaces']:
        raw = reader.read(original, 'source-capture/' + name + '.json')
        if sha(raw) != before['source_sha256'][name]:
            raise ValueError('source digest mismatch')
        inputs[name] = json.loads(raw)
    fresh, files = prepare_bundle(inputs['spaces'], inputs['articles'],
                                  inventory_spaces=inputs['retained_spaces'])
    # New report guards must not alter the original draft content or sidecars.
    if {name: sha(raw) for name, raw in files.items()} != before['file_sha256']:
        raise ValueError('draft file transformation changed')
    for name, raw in files.items():
        if reader.read(original, name) != raw:
            raise ValueError('original draft file drift')
    bodies = {}
    for article in fresh['articles']:
        if article['body_file']:
            raw = reader.read(original, article['body_file'])
            if sha(raw) != article['body_sha256']:
                raise ValueError('body sidecar drift')
            bodies[article['id']] = raw
    prepared_pages = 0
    for space in visual['spaces']:
        for page in space['articles']:
            if page['file']:
                raw = reader.read(prepared, page['file'])
                if sha(raw) != visual['file_sha256'][page['file']]:
                    raise ValueError('prepared page drift')
                if raw.split(b'\n---\n', 1)[1] != bodies[page['id']]:
                    raise ValueError('prepared source body changed')
                prepared_pages += 1
    details = []
    for article in fresh['articles']:
        detail = {'id': article['id'], 'body_present': article['id'] in bodies,
                  'references': [r for r in article['references'] if r['kind'] != 'visual'],
                  'review_flags': article['review_flags']}
        if 'scrap_relation' in article:
            detail['scrap_relation'] = article['scrap_relation']
        if detail['body_present']:
            detail['rendering'], _ = audit_body(bodies[article['id']].decode())
        details.append(detail)
    rendered = [a['rendering'] for a in details if a['body_present']]
    refs = [r for a in details for r in a['references']]
    summary = {
        'schema': 'coulomb.bubble-body-resource-proof.v1',
        'source_report_sha256': sha(raw_report), 'visual_report_sha256': sha(visual_raw),
        'source_sha256': before['source_sha256'],
        'captured_articles': len(details), 'source_bodies_verified': len(bodies),
        'missing_bodies': len(details) - len(bodies), 'prepared_pages_verified': prepared_pages,
        'original_draft_files_recompiled_unchanged': len(files),
        'body_reference_inventory_entries': len(refs),
        'body_reference_records': sum(bool(a['references']) for a in details),
        'body_reference_classifications': dict(Counter(r['classification'] for r in refs)),
        'reference_occurrences_before': sum(r['reference_occurrences_before'] for r in rendered),
        'reference_occurrences_after': sum(r['reference_occurrences_after'] for r in rendered),
        'records_with_changed_reference_attributes': sum(
            not r['reference_attributes_preserved'] for r in rendered),
        'unresolved_scrap_relation_records': sum(
            a.get('scrap_relation', {}).get('state') == 'unresolved' for a in details),
        'attachment_flag_records': sum('attachment_records_require_capture' in a['review_flags']
                                       for a in details),
        'live_destinations_checked': False, 'original_bubble_render_compared': False,
        'productive_fidelity_accepted': False, 'production_imported': False,
        'network_requests': 0,
    }
    return summary, {'summary': summary, 'records': details,
                     'fresh_space_blockers': [{'id': s['id'], 'blockers': s['blockers']}
                                              for s in fresh['spaces']]}


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('original', type=Path)
    parser.add_argument('prepared', type=Path)
    parser.add_argument('--output', type=Path, required=True, help='New private JSON outside repo')
    args = parser.parse_args()
    try:
        out = args.output
        if (out.exists() or any(p.is_symlink() for p in (out, *out.parents))
                or any(out.resolve().is_relative_to(p.resolve())
                       for p in [ROOT, args.original, args.prepared])):
            raise ValueError('separate new output required')
        info = out.parent.stat()
        if info.st_uid != os.getuid() or info.st_mode & 0o077:
            raise ValueError('private owned output parent required')
        summary, private = verify(args.original, args.prepared)
        raw = json.dumps(private, indent=2).encode()
        with os.fdopen(os.open(out, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600), 'wb') as f:
            f.write(raw)
            f.flush()
            os.fsync(f.fileno())
        summary['private_report_sha256'] = sha(raw)
        print(json.dumps(summary))
    except Exception:
        print('Body resource proof refused; no completeness or acceptance inferred.',
              file=sys.stderr)
        return 1
    return 0


if __name__ == '__main__':
    raise SystemExit(main())