net-kingdom/tests/test_keycape_attempt_receipt_verifier.py

"""Synthetic tests for the independent KeyCape attempt receipt verifier; no cluster, no values."""
import copy
from datetime import datetime, timezone
import hashlib
import importlib.util
import json
from pathlib import Path
import tempfile
import unittest

SOURCE = Path(__file__).parents[1] / 'tools/bootstrap-recovery/verify-keycape-attempt-receipt.py'
spec = importlib.util.spec_from_file_location('verify_receipt', SOURCE)
v = importlib.util.module_from_spec(spec)
spec.loader.exec_module(v)

CIPHERTEXT = 'c' * 64


class Verifier(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.root = Path(self.tmp.name)
        hashes = {}
        for name in v.SOURCE_FILES:
            (self.root / name).write_bytes(f'synthetic {name}\n'.encode())
            hashes[name] = hashlib.sha256((self.root / name).read_bytes()).hexdigest()
        done = {'status': 'absence_verified'}
        self.receipt = {
            'schema': v.SCHEMA, 'status': v.PASSED, 'provision_phase': 'complete',
            **{k: True for k in v.REQUIRED_TRUE}, **{k: False for k in v.REQUIRED_FALSE},
            **{k: [dict(done)] for k in v.OBLIGATIONS}, 'plaintext_cleanup_obligation': dict(done),
            'ciphertext_sha256': CIPHERTEXT, 'public_signing_spki_sha256': 'a' * 64,
            'source_files_sha256': hashes, 'session_source_sha256': hashes['isolated-keycape-session.py'],
            'finished_at': '2026-10-24T10:00:00+00:00'}
        self.transfer = {'ciphertext_sha256': CIPHERTEXT}

    def tearDown(self):
        self.tmp.cleanup()

    def check(self, receipt=None, transfer=None, not_after=None):
        raw = json.dumps(self.receipt if receipt is None else receipt).encode()
        return v.verify(raw, self.transfer if transfer is None else transfer, self.root, not_after=not_after)

    def mutated(self, **changes):
        r = copy.deepcopy(self.receipt)
        r.update(changes)
        return r

    def test_complete_receipt_verifies(self):
        self.assertEqual(self.check(), [])

    def test_refuses_failed_or_partial_outcomes(self):
        cases = [self.mutated(status='failed'), self.mutated(status='bounded_cleanup_uncertain'),
                 self.mutated(provision_phase='issuer_launch'), self.mutated(schema='other'),
                 self.mutated(error='isolated_keycape_failed_private_diagnostics_suppressed')]
        cases += [self.mutated(**{k: False}) for k in v.REQUIRED_TRUE]
        missing = copy.deepcopy(self.receipt)
        del missing['denied_scopes_refused']
        cases.append(missing)
        for receipt in cases:
            with self.subTest(receipt={k: receipt.get(k) for k in ('status', 'provision_phase')}):
                self.assertTrue(self.check(receipt))

    def test_refuses_overclaims_and_uncertainty(self):
        for key in v.REQUIRED_FALSE:
            with self.subTest(key=key):
                self.assertIn(f'{key} is not false', self.check(self.mutated(**{key: True})))

    def test_refuses_open_cleanup_obligations(self):
        open_row = [{'status': 'registered_before_plaintext'}]
        for key in v.OBLIGATIONS:
            with self.subTest(key=key):
                self.assertTrue(self.check(self.mutated(**{key: open_row})))
        self.assertTrue(self.check(self.mutated(plaintext_cleanup_obligation={'status': 'pending'})))

    def test_binds_ciphertext_to_transfer_receipt(self):
        self.assertIn('ciphertext_sha256 differs from the transfer receipt',
                      self.check(transfer={'ciphertext_sha256': 'd' * 64}))
        self.assertIn('transfer receipt lacks ciphertext_sha256', self.check(transfer={}))

    def test_recomputes_source_from_checkout(self):
        (self.root / 'isolated-keycape.py').write_bytes(b'changed\n')
        self.assertIn('source differs from checkout: isolated-keycape.py', self.check())

    def test_session_source_must_match(self):
        self.assertIn('session_source_sha256 differs', self.check(self.mutated(session_source_sha256='0' * 64)))

    def test_admission_window(self):
        early = datetime(2026, 10, 24, 9, 0, tzinfo=timezone.utc)
        late = datetime(2026, 10, 24, 11, 0, tzinfo=timezone.utc)
        self.assertIn('attempt finished after the admission window', self.check(not_after=early))
        self.assertEqual(self.check(not_after=late), [])

    def test_secret_shaped_content_is_refused(self):
        for value in ('-----BEGIN PRIVATE KEY-----', 'AGE-SECRET-KEY-1QQQQ',
                      'eyJhbGciOiJSUzI1NiJ9.eyJzdWIiOiJzZXJ2aWNlIn0.sig', 'A' * 200):
            with self.subTest(value=value[:12]):
                findings = self.check(self.mutated(note=value))
                self.assertIn('secret-shaped content present; do not retain or publish this receipt', findings)

    def test_cli_exit_codes(self):
        receipt, transfer = self.root / 'recovery.json', self.root / 'transfer.json'
        receipt.write_text(json.dumps(self.receipt))
        transfer.write_text(json.dumps(self.transfer))
        args = ['--receipt', str(receipt), '--transfer-receipt', str(transfer), '--source-root', str(self.root)]
        self.assertEqual(v.main(args), 0)
        receipt.write_text(json.dumps(self.mutated(status='failed')))
        self.assertEqual(v.main(args), 1)


if __name__ == '__main__':
    unittest.main()