net-kingdom/tools/bootstrap-recovery/verify-keycape-attempt-receipt.py

#!/usr/bin/env python3
"""Independent verifier for an isolated KeyCape attempt receipt (NK-WP-0049-T33/T18).

Reads the safe receipt written by run-isolated-keycape.sh, the public transfer
receipt and a checkout of the reviewed source. It imports nothing from the
session code: source hashes are recomputed from the checkout, the ciphertext
binding comes from the transfer receipt, and the receipt itself is scanned for
secret-shaped content before anyone retains or publishes it. A pass means the
receipt is internally complete and value-safe. It is not consumer continuity
(T18), retained-factor MFA (T19) or owner acceptance.
"""
import argparse
import hashlib
import json
import re
import sys
from datetime import datetime
from pathlib import Path

SCHEMA = 'netkingdom.isolated-keycape-recovery.v1'
PASSED = 'isolated_keycape_native_registration_checked'
SOURCE_FILES = ('isolated-keycape.py', 'isolated-keycape-session.py', 'run-isolated-keycape.sh',
                'isolated-lldap.py', 'identity-bundle.py', 'verify-identity.py',
                'retrieve-identity-offline.py', 'identity-offsite.py')
REQUIRED_TRUE = ('all_manifest_hashes_verified', 'native_service_registration_verified',
                 'wrong_unrelated_secret_explicitly_refused', 'denied_scopes_refused',
                 'original_config_registry_policy_receipts_equal',
                 'original_https_issuer_and_trust_preserved', 'original_public_signing_key_in_live_jwks',
                 'temporary_plaintext_cleanup_verified', 'container_cleanup_verified',
                 'bounded_cleanup_reaper_started')
REQUIRED_FALSE = ('production_contacted', 'secret_values_exported', 'in_flight_daemon_request_uncertain',
                  'downstream_consumer_reconnect_proven', 'human_mfa_proven')
OBLIGATIONS = ('container_cleanup_obligations', 'tmpfs_volume_cleanup_obligations',
               'public_image_probe_cleanup_obligations')
SECRET_SHAPES = (re.compile(r'-----BEGIN [A-Z ]*(PRIVATE KEY|CERTIFICATE)'),
                 re.compile(r'AGE-SECRET-KEY-1'),
                 re.compile(r'eyJ[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}\.'),
                 re.compile(r'[A-Za-z0-9+/]{120,}={0,2}'))
SHA256 = re.compile(r'^[a-f0-9]{64}$')


def sha256(path):
    return hashlib.sha256(path.read_bytes()).hexdigest()


def verify(raw, transfer, source_root, *, not_after=None):
    """Return a list of findings; empty means the receipt verifies."""
    findings = []
    text = raw.decode('utf-8', errors='replace')
    for shape in SECRET_SHAPES:
        if shape.search(text):
            findings.append('secret-shaped content present; do not retain or publish this receipt')
            break
    try:
        receipt = json.loads(raw)
    except ValueError:
        return findings + ['receipt is not JSON']
    if not isinstance(receipt, dict):
        return findings + ['receipt is not an object']
    if receipt.get('schema') != SCHEMA:
        findings.append('schema differs')
    if receipt.get('status') != PASSED:
        findings.append(f"status is {receipt.get('status')!r}, not {PASSED}")
    findings += [f'{k} is not true' for k in REQUIRED_TRUE if receipt.get(k) is not True]
    findings += [f'{k} is not false' for k in REQUIRED_FALSE if receipt.get(k) is not False]
    for key in OBLIGATIONS:
        rows = receipt.get(key, [])
        if not isinstance(rows, list) or any(not isinstance(r, dict) or r.get('status') != 'absence_verified'
                                             for r in rows):
            findings.append(f'{key} not all absence_verified')
    plaintext = receipt.get('plaintext_cleanup_obligation')
    if not isinstance(plaintext, dict) or plaintext.get('status') != 'absence_verified':
        findings.append('plaintext_cleanup_obligation not absence_verified')
    if receipt.get('provision_phase') != 'complete':
        findings.append('provision_phase is not complete')
    if 'error' in receipt:
        findings.append('receipt records an error')
    expected_ct = transfer.get('ciphertext_sha256') if isinstance(transfer, dict) else None
    if not isinstance(expected_ct, str) or not SHA256.match(expected_ct):
        findings.append('transfer receipt lacks ciphertext_sha256')
    elif receipt.get('ciphertext_sha256') != expected_ct:
        findings.append('ciphertext_sha256 differs from the transfer receipt')
    spki = receipt.get('public_signing_spki_sha256')
    if not isinstance(spki, str) or not SHA256.match(spki):
        findings.append('public_signing_spki_sha256 missing')
    recorded = receipt.get('source_files_sha256')
    if not isinstance(recorded, dict) or set(recorded) != set(SOURCE_FILES):
        findings.append('source_files_sha256 does not cover the reviewed source set')
    else:
        for name in SOURCE_FILES:
            path = source_root / name
            if not path.is_file() or sha256(path) != recorded[name]:
                findings.append(f'source differs from checkout: {name}')
        if receipt.get('session_source_sha256') != recorded.get('isolated-keycape-session.py'):
            findings.append('session_source_sha256 differs')
    finished = receipt.get('finished_at')
    try:
        finished_at = datetime.fromisoformat(finished)
    except (TypeError, ValueError):
        findings.append('finished_at missing or malformed')
    else:
        if not_after is not None and finished_at > not_after:
            findings.append('attempt finished after the admission window')
    return findings


def main(argv=None):
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument('--receipt', required=True, type=Path)
    p.add_argument('--transfer-receipt', required=True, type=Path)
    p.add_argument('--source-root', required=True, type=Path,
                   help='tools/bootstrap-recovery of a checkout at the reviewed commit')
    p.add_argument('--not-after', help='admission window end, ISO 8601 with offset')
    args = p.parse_args(argv)
    not_after = datetime.fromisoformat(args.not_after) if args.not_after else None
    if not_after is not None and not_after.tzinfo is None:
        p.error('--not-after needs a UTC offset')
    findings = verify(args.receipt.read_bytes(), json.loads(args.transfer_receipt.read_text()),
                      args.source_root, not_after=not_after)
    print(json.dumps({'schema': 'netkingdom.keycape-attempt-receipt-verification.v1',
                      'receipt_sha256': sha256(args.receipt),
                      'verdict': 'verified' if not findings else 'refused', 'findings': findings,
                      'claims_not_made': ['consumer continuity (T18)', 'retained-factor MFA (T19)',
                                          'owner acceptance']}, indent=2))
    return 0 if not findings else 1


if __name__ == '__main__':
    sys.exit(main())