railiance-platform/tests/test_archive_recovery_evidence.py

"""Full primary recovery requires a complete, ordered, hash-bound receipt chain."""
import copy
from datetime import datetime, timezone, timedelta
import hashlib
import json
from pathlib import Path
import sys

import pytest

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'scripts'))
from recovery_evidence import recovery_signals, ROOT
from service_assurance import evaluate

NOW = datetime(2026, 9, 27, 12, tzinfo=timezone.utc)
DESTINATION = 's3://railiance-platform-pg-backup/platform-pg/application-archives/forgejo/fixture.zip.age'


def chain():
    transfer = dict(schema='platform.forgejo-primary-archive.v1',
        status='primary_fetched_pending_application_restore', stage='transfer_verified',
        started_at='2026-09-27T10:00:00Z', finished_at='2026-09-27T10:01:00Z',
        archive_profile='full', destination=DESTINATION, ciphertext_sha256='a'*64,
        ciphertext_bytes=123, uploaded_bytes=123, downloaded_bytes=123,
        multipart_completed=True, version_pinned=True, download_hash_matches=True)
    decryption = dict(schema='platform.forgejo-primary-decryption.v1',
        status='primary_fetched_pending_application_restore',
        started_at='2026-09-27T10:02:00Z', finished_at='2026-09-27T10:03:00Z',
        archive_profile='full', destination=DESTINATION, ciphertext_sha256='a'*64,
        ciphertext_bytes=123, decrypted=True, download_hash_matches=True,
        plaintext_sha256='b'*64)
    restore = dict(schema='platform.forgejo-isolated-restore.v1', status='restored',
        started_at='2026-09-27T10:04:00Z', finished_at='2026-09-27T10:05:00Z',
        archive_profile='full', source_provider='Scaleway', stage='package_blob_recovery',
        offsite_artifact=DESTINATION, ciphertext_sha256='a'*64, database_import=True,
        application_health=True, cleanup=True, repositories_verified=['fixture/repo'],
        package_blobs_verified=3, database_counts={'package_blobs': 3})
    return transfer, decryption, restore


def write_chain(root, receipts, broken_link=None):
    def write(name, receipt):
        relative = f'docs/evidence/{name}.json'
        path = root / relative
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(json.dumps(receipt))
        return {'path': relative, 'sha256': hashlib.sha256(path.read_bytes()).hexdigest()}
    transfer, decryption, restore = copy.deepcopy(receipts)
    source = write('transfer', transfer)
    decryption['transfer_receipt_sha256'] = source['sha256'] if broken_link != 'transfer' else '0'*64
    decrypted = write('decryption', decryption)
    restore['transfer_receipt_sha256'] = decrypted['sha256'] if broken_link != 'decryption' else '0'*64
    recovered = write('restore', restore)
    index = {'schema': 'railiance-platform.recovery-evidence.v1', 'receipts': [
        {'signal': 'offsite.upload', **source},
        {'signal': 'offsite.restore', **recovered, 'decryption': decrypted, 'transfer': source}]}
    (root / 'assurance').mkdir(exist_ok=True)
    (root / 'assurance/recovery-evidence.json').write_text(json.dumps(index))
    return index


def test_complete_chain_preserves_times_and_expires(tmp_path):
    write_chain(tmp_path, chain())
    signals = recovery_signals(NOW, tmp_path)
    assert signals['offsite.upload'] == {'result': 'pass', 'observed_at': '2026-09-27T10:01:00Z'}
    assert signals['offsite.restore'] == {'result': 'pass', 'observed_at': '2026-09-27T10:05:00Z'}
    later = NOW + timedelta(days=31)
    contract = {'cluster_uid': 'fixture', 'capture_max_age_seconds': 900,
        'signals': {s: {'owner': 'platform', 'max_age_seconds': 2592000} for s in signals}}
    report = evaluate(contract, {'schema': 'railiance-platform.observation.v1',
        'cluster_uid': 'fixture', 'captured_at': later.isoformat(),
        'signals': recovery_signals(later, tmp_path)}, later)
    assert all(s['state'] == 'stale' for s in report['signals'].values())


@pytest.mark.parametrize('index,key,value', [
    (0, 'version_pinned', False), (0, 'download_hash_matches', False),
    (0, 'downloaded_bytes', 122), (0, 'ciphertext_bytes', True),
    (0, 'destination', 's3://other/fixture'), (0, 'archive_profile', 'essentials'),
    (0, 'finished_at', '2027-01-01T00:00:00Z'),
    (1, 'archive_profile', 'essentials'), (1, 'decrypted', False),
    (1, 'destination', DESTINATION + 'other'), (1, 'ciphertext_sha256', 'c'*64),
    (1, 'started_at', '2026-09-27T10:00:30Z'),
    (1, 'schema', 'platform.forgejo-primary-archive.v1'),
    (2, 'status', 'failed'), (2, 'cleanup', False), (2, 'application_health', False),
    (2, 'database_import', False), (2, 'archive_profile', 'essentials'),
    (2, 'source_provider', 'Nextcloud'), (2, 'offsite_artifact', DESTINATION + 'other'),
    (2, 'package_blobs_verified', 2), (2, 'package_blobs_verified', True),
    (2, 'repositories_verified', []), (2, 'finished_at', '2026-09-27T10:05:00'),
    (2, 'started_at', '2026-09-27T10:02:30Z'),
])
def test_incomplete_or_wrong_recovery_never_passes(tmp_path, index, key, value):
    receipts = chain()
    receipts[index][key] = value
    write_chain(tmp_path, receipts)
    signals = recovery_signals(NOW, tmp_path)
    assert signals['offsite.restore']['result'] == 'unavailable'
    if index == 0:
        assert signals['offsite.upload']['result'] == 'unavailable'


@pytest.mark.parametrize('link', ['transfer', 'decryption'])
def test_provenance_links_must_match_exact_bytes(tmp_path, link):
    write_chain(tmp_path, chain(), broken_link=link)
    assert recovery_signals(NOW, tmp_path)['offsite.restore']['result'] == 'unavailable'


@pytest.mark.parametrize('case', ['drift', 'missing', 'escape', 'missing_time', 'malformed'])
def test_untrusted_files_are_unavailable(tmp_path, case):
    index = write_chain(tmp_path, chain())
    entry = index['receipts'][1]
    path = tmp_path / entry['path']
    if case == 'drift': path.write_text('{}')
    if case == 'missing': path.unlink()
    if case == 'escape': entry['path'] = '/tmp/outside-recovery-evidence.json'
    if case in ('missing_time', 'malformed'):
        receipt = json.loads(path.read_text())
        if case == 'missing_time': del receipt['finished_at']
        else: receipt = []
        path.write_text(json.dumps(receipt))
        entry['sha256'] = hashlib.sha256(path.read_bytes()).hexdigest()
    (tmp_path / 'assurance/recovery-evidence.json').write_text(json.dumps(index))
    assert recovery_signals(NOW, tmp_path)['offsite.restore']['result'] == 'unavailable'


def test_historical_undated_receipt_is_not_freshened(tmp_path):
    receipts = list(chain())
    receipts[2] = json.loads((ROOT / 'docs/evidence/RPF-WP-0038-primary-archive-restore-2026-09-06.json').read_text())
    write_chain(tmp_path, receipts)
    assert recovery_signals(NOW, tmp_path)['offsite.restore']['result'] == 'unavailable'