email-connect/scripts/rehearse-ledger.py

#!/usr/bin/env python3
"""Attended recovered-ledger rehearsal in the cached receiver image, no network."""

from __future__ import annotations

import argparse
import json
import re
import secrets
import subprocess
import time
from pathlib import Path

from ledger_archive import (
    Refused,
    checkpoint,
    digest,
    private_directory,
    private_file,
    validate_capture,
)

IMAGE = (
    "forgejo.coulomb.social/coulomb/email-connect@sha256:"
    "eb3e7b68d21209e9eb71b928a5b0bc987e8e3164f21dca4ff712f946d0c103f2"
)

CODE = """import io,json,os,sys
from pathlib import Path
sys.path.insert(0,'/opt/ledger-rehearsal')
from ledger_archive import fingerprint
from email_connect.transactional import SQLiteDeliveryStore,TransactionalApplication,ProviderError
assert os.getuid()==10001
path=Path('/tmp/recovered.sqlite3')
os.umask(0o077)
path.write_bytes(sys.stdin.buffer.read())
path.chmod(0o600)
before=fingerprint(path)
store=SQLiteDeliveryStore(str(path))
assert fingerprint(path)==before
class ForbiddenProvider:
    def send(self,*args):raise AssertionError('recovery_provider_call_forbidden')
app=TransactionalApplication(store,ForbiddenProvider(),'synthetic-recovery-token','https://issuer.example.test')
checks={'accepted_verifications':0,'unknown_verifications':0,'invalidated_handles':0,
        'suppressions':0,'accepted_deliveries':0,'unknown_sends':0}
caller_column="v.caller_id" if "caller_id" in {r[1] for r in store.db.execute("PRAGMA table_info(verifications)")} else "''"
query=("SELECT v.registration_id,v.email,v.username,v.client_id,v.tenant,v.display_name,a.state,"+caller_column+" "
       "FROM verifications v JOIN verification_attempts a ON a.request_id=v.request_id "
       "WHERE a.state IN ('accepted','in_flight')")
for row in store.db.execute(query).fetchall():
    app=TransactionalApplication(store,ForbiddenProvider(),'synthetic-recovery-token',
        'https://issuer.example.test',**({'verification_namespace':row[7]} if row[7] else {}))
    payload=dict(zip(['registration_id','normalized_email','preferred_username','client_id','tenant','display_name'],row[:6]))
    payload['correlation_id']='synthetic-recovery-probe'
    raw=json.dumps(payload).encode();response={}
    body=json.loads(b''.join(app({'PATH_INFO':'/v1/registration-verifications','REQUEST_METHOD':'POST',
        'CONTENT_LENGTH':str(len(raw)),'wsgi.input':io.BytesIO(raw),
        'HTTP_AUTHORIZATION':'Bearer synthetic-recovery-token'},
        lambda status,headers:response.update(status=status))))
    if row[6]=='accepted':
        assert response['status'].startswith('202') and body['accepted']
        checks['accepted_verifications']+=1
    else:
        assert response['status'].startswith('422') and body['error']=='delivery_outcome_unknown'
        checks['unknown_verifications']+=1
query=("SELECT handle_hash FROM verifications "
       "WHERE consumed_at IS NOT NULL OR canceled_at IS NOT NULL")
for row in store.db.execute(query).fetchall():
    try:store.consume_verification(row[0]);raise AssertionError('invalid_handle_reactivated')
    except ValueError as error:assert str(error)=='verification_invalid'
    checks['invalidated_handles']+=1
for row in store.db.execute('SELECT email FROM suppressions').fetchall():
    assert store.is_suppressed(row[0]);checks['suppressions']+=1
query=("SELECT d.event_id,a.fingerprint FROM deliveries d "
       "LEFT JOIN send_attempts a ON a.event_id=d.event_id")
for row in store.db.execute(query).fetchall():
    assert store.delivery_status(row[0])['state']=='provider_accepted'
    assert store.reserve_send(row[0],row[1] or 'synthetic-legacy-probe') is False
    checks['accepted_deliveries']+=1
query=("SELECT event_id,fingerprint FROM send_attempts WHERE state='in_flight' "
       "AND event_id NOT IN (SELECT event_id FROM deliveries)")
for row in store.db.execute(query).fetchall():
    assert store.delivery_status(row[0])['state']=='unknown'
    try:store.reserve_send(row[0],row[1]);raise AssertionError('unknown_send_retried')
    except ProviderError as error:assert error.code=='delivery_outcome_unknown'
    checks['unknown_sends']+=1
store.db.close()
assert fingerprint(path)==before
print(json.dumps({'status':'isolated_ledger_verified','ledger_fingerprint':before,
                  'checks':checks,'mail_sent':False,'uid':os.getuid()}))
"""


def recovery_image(offline_binding=None):
    if offline_binding is None:
        return IMAGE
    private_file(offline_binding)
    binding = json.loads(offline_binding.read_text())
    image_id = binding.get("image_id", "")
    if (binding.get("schema") != "email-connect.ledger-recovery-kit.v1"
            or binding.get("image") != IMAGE
            or not isinstance(image_id, str)
            or not re.fullmatch(r"sha256:[a-f0-9]{64}", image_id)):
        raise Refused("offline_image_binding_required")
    archive = offline_binding.parent / "receiver-image.tar"
    private_file(archive)
    expected = binding["files"]["receiver-image.tar"]
    if archive.stat().st_size != expected["bytes"] or digest(archive) != expected["sha256"]:
        raise Refused("offline_image_archive_changed")
    return image_id


def rehearse(recovered, receipt_path, offline_binding=None):
    private_directory(recovered)
    private_directory(receipt_path.parent)
    if receipt_path.exists() or receipt_path.is_symlink():
        raise Refused("fresh_receipt_required")
    capture = validate_capture(recovered)
    private_file(recovered / "unpack-receipt.json")
    unpacked = json.loads((recovered / "unpack-receipt.json").read_text())
    if (
        capture["image"] != IMAGE
        or unpacked.get("status") != "decrypted_ledger_verified"
        or unpacked.get("ledger_fingerprint") != capture["ledger_fingerprint"]
    ):
        raise Refused("pinned_recovered_ledger_required")
    # Never pull a recovery image from the production forge during this rehearsal.
    selected_image = recovery_image(offline_binding)
    inspected = subprocess.run(
        ["docker", "image", "inspect", selected_image], capture_output=True, timeout=30
    )
    if inspected.returncode:
        raise Refused("cached_pinned_recovery_image_required")
    started = time.monotonic()
    module = Path(__file__).with_name("ledger_archive.py").resolve()
    name = "email-ledger-rehearsal-" + secrets.token_hex(8)
    try:
        with (recovered / "database.sqlite3").open("rb") as source:
            result = subprocess.run(
                [
                    "docker",
                    "run",
                    "--rm",
                    "--pull=never",
                    "--name",
                    name,
                    "-i",
                    "--network",
                    "none",
                    "--read-only",
                    "--user",
                    "10001:10001",
                    "--cap-drop",
                    "ALL",
                    "--security-opt",
                    "no-new-privileges",
                    "--memory",
                    "512m",
                    "--cpus",
                    "0.5",
                    "--tmpfs",
                    "/tmp:rw,noexec,nosuid,size=160m,uid=10001,gid=10001,mode=0700",
                    "--mount",
                    "type=bind,source="
                    + str(module)
                    + ",target=/opt/ledger-rehearsal/ledger_archive.py,readonly",
                    "-e",
                    "PYTHONDONTWRITEBYTECODE=1",
                    "--entrypoint",
                    "python",
                    selected_image,
                    "-c",
                    CODE,
                ],
                stdin=source,
                capture_output=True,
                timeout=120,
            )
    finally:
        subprocess.run(["docker", "rm", "-f", name], capture_output=True, timeout=30)
    if result.returncode:
        raise Refused("isolated_rehearsal_failed")
    verification = json.loads(result.stdout)
    if verification.get("ledger_fingerprint") != capture["ledger_fingerprint"]:
        raise Refused("isolated_fingerprint_mismatch")
    receipt = dict(
        unpacked,
        status="application_restored_verified",
        application_restore_verified=True,
        recovery_image=IMAGE,
        network="disabled",
        production_restored=False,
        checks=verification["checks"],
        uid=verification["uid"],
        mail_sent=False,
        evidence_ceiling="isolated_ledger_rehearsal",
        offline_image_binding_verified=offline_binding is not None,
        elapsed_seconds=round(time.monotonic() - started, 2),
    )
    checkpoint(receipt_path, receipt)
    return receipt


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--recovered", required=True, type=Path)
    parser.add_argument("--receipt", required=True, type=Path)
    parser.add_argument("--execute", action="store_true")
    parser.add_argument("--offline-image-binding", type=Path)
    args = parser.parse_args()
    if not args.execute:
        print(
            json.dumps(
                {"status": "reviewed", "container_started": False, "production_restored": False}
            )
        )
        return 0
    print(json.dumps(rehearse(args.recovered, args.receipt, args.offline_image_binding)))
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except Exception as error:
        print(json.dumps({"status": "failed", "failure_type": type(error).__name__}))
        raise SystemExit(1)