net-kingdom/tools/bootstrap-recovery/isolated-keycape-session.py

#!/usr/bin/env python3
"""Owner-run T18 session assembly using the existing archive/containment pipeline.

Candidate only: native pinned-image tests and review are required before use.
Shared Docker local-driver tmpfs is required because Docker cp does not support
private tmpfs/mount paths, and the original KeyCape image is distroless.
"""
import argparse
import base64
from datetime import datetime, timezone
import hashlib
import importlib.util
import json
import os
from pathlib import Path
import signal
import re
import stat
import select
import subprocess
import sys
import tempfile
import time
import uuid

HERE=Path(__file__).resolve().parent

def load(name):
    spec=importlib.util.spec_from_file_location(name,HERE/(name+'.py'))
    module=importlib.util.module_from_spec(spec);spec.loader.exec_module(module);return module

adapter=load('isolated-keycape')
local=load('isolated-lldap')
bundle=local.bundle
# Bind the reused preflight itself, not only this module's direct Docker calls.
local.command=adapter.command
_original_preflight=local.preflight
def preflight():
    contexts=json.loads(adapter.command(['docker','context','inspect']))
    if len(contexts)!=1 or contexts[0]['Endpoints']['docker']['Host']!=adapter.ENDPOINT:raise adapter.Refused()
    result=_original_preflight();result['docker_endpoint']=adapter.ENDPOINT;return result
local.preflight=preflight
WORKER=local.CLIENT
STAGING_ROOT=Path('/dev/shm')
RUNROOTS=('/etc/keycape','/var/lib/keycape-policy','/etc/keycape-factor','/recovery-tls')
VOLUME_OPTIONS={'type':'tmpfs','device':'tmpfs','o':'size=33554432,uid=1000,gid=1000,mode=0700,nosuid,nodev,noexec'}

IMPORT=r'''
import base64,json,os,sys
from pathlib import Path
os.umask(0o077)
p=json.load(sys.stdin)
allowed=('/etc/keycape/','/var/lib/keycape-policy/','/etc/keycape-factor/','/recovery-tls/')
for name,encoded in p.items():
    f=Path(name)
    if not name.startswith(allowed) or '..' in f.parts or any(x.is_symlink() for x in (f,*f.parents)):raise ValueError()
    f.parent.mkdir(mode=0o700,parents=True,exist_ok=True)
    fd=os.open(f,os.O_WRONLY|os.O_CREAT|os.O_EXCL|os.O_NOFOLLOW,0o600)
    with os.fdopen(fd,'wb') as out:out.write(base64.b64decode(encoded,validate=True))
'''

# Small transport inside the already pinned helper; it has only loopback routes.
# No token handling, auth bypass, dependency imitation, reset or notifier path.
TLS=r'''
import select,socket,socketserver,ssl,threading
context=ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
context.minimum_version=ssl.TLSVersion.TLSv1_2
context.load_cert_chain('/recovery-tls/server.pem','/recovery-tls/server-key.pem')
class Handler(socketserver.BaseRequestHandler):
    def handle(self):
        try:
            with context.wrap_socket(self.request,server_side=True) as incoming:
                with socket.create_connection(('127.0.0.1',8080),timeout=10) as outgoing:
                    incoming.settimeout(15);outgoing.settimeout(15)
                    while True:
                        ready,_,_=select.select([incoming,outgoing],[],[],15)
                        if not ready:return
                        for source in ready:
                            data=source.recv(65536)
                            if not data:return
                            (outgoing if source is incoming else incoming).sendall(data)
        except Exception:pass
class Server(socketserver.ThreadingTCPServer):
    allow_reuse_address=True
    daemon_threads=True
with Server(('127.0.0.1',443),Handler) as server:server.serve_forever()
'''


def run(argv,**kwargs):return adapter.command(argv,**kwargs)


def plan_session():
    run_id=uuid.uuid4().hex
    stem='nk49-kc-'+run_id
    worker=stem+'-transport';issuer=stem+'-issuer'
    volumes={stem+'-v'+str(i):{'target':root,'options':dict(VOLUME_OPTIONS)} for i,root in enumerate(RUNROOTS)}
    return {'run_id':run_id,'containers':[worker,issuer], 'worker':worker,'issuer':issuer,
            'images':{worker:WORKER,issuer:adapter.IMAGE},'volumes':volumes,
            'public_image_probe':stem+'-image-probe','docker_endpoint':adapter.ENDPOINT}


def verify_volume(name,spec,run_id):
    obj=json.loads(run(['docker','volume','inspect',name]))
    if (len(obj)!=1 or obj[0]['Name']!=name or obj[0]['Driver']!='local' or obj[0].get('Scope')!='local'
            or obj[0].get('Options')!=VOLUME_OPTIONS or spec['options']!=VOLUME_OPTIONS
            or obj[0].get('Labels',{}).get(adapter.LABEL)!=run_id):raise adapter.Refused()


def base_args(session,name,network):
    return ['docker','run','--detach','--pull=never','--name',name,'--network',network,
       '--label',adapter.LABEL+'='+session['run_id'],'--read-only','--cap-drop','ALL',
       '--security-opt','no-new-privileges','--user','1000:1000','--memory','512m','--cpus','1',
       '--pids-limit','64','--log-driver','none','--restart','no',
       '--tmpfs','/tmp:rw,nosuid,nodev,noexec,size=16777216,uid=1000,gid=1000,mode=0700']


def mounts(session,readonly=False):
    args=[]
    for name,spec in session['volumes'].items():
        args+=['--mount','type=volume,source='+name+',target='+spec['target']+(',readonly' if readonly and spec['target']!='/var/lib/keycape-policy' else '')]
    return args


PROVISION_PHASES=(
    'volume_creation_and_verification','worker_launch','worker_containment',
    'private_import','issuer_launch','issuer_containment','tls_transport_launch','complete')


def provision(session,payload,tls,issuer,phase=None):
    """Inspect containment and driver options before the first private import."""
    from urllib.parse import urlsplit
    address=urlsplit(issuer)
    if address.scheme!='https' or not address.hostname or address.port not in (None,443) or address.path not in ('','/'):
        raise adapter.Refused()
    if phase is None:phase=lambda value:None
    phase('volume_creation_and_verification')
    for name,spec in session['volumes'].items():
        run(['docker','volume','create','--driver','local','--label',adapter.LABEL+'='+session['run_id'],
             '--opt','type=tmpfs','--opt','device=tmpfs','--opt','o='+VOLUME_OPTIONS['o'],name])
        verify_volume(name,spec,session['run_id'])
    phase('worker_launch')
    args=base_args(session,session['worker'],'none')+mounts(session)
    # This pinned helper declares an image VOLUME here. Mask it explicitly so
    # Docker cannot allocate an unregistered disk-backed anonymous volume.
    args+=['--tmpfs','/privacyidea/etc/persistent:rw,nosuid,nodev,noexec,size=16777216,uid=1000,gid=1000,mode=0700']
    args+=['--sysctl','net.ipv4.ip_unprivileged_port_start=0','--add-host',address.hostname+':127.0.0.1',
           '--entrypoint','python3',WORKER,'-c','import time;time.sleep(1800)']
    run(args)
    # Check helper and every private filesystem before touching the ciphertext's
    # private contents. Issuer is created only after originals are present.
    phase('worker_containment')
    adapter.inspect_session({**session,'containers':[session['worker']],'issuer':session['worker']},worker_only=True)
    phase('private_import')
    files={**payload['files'],**{k:base64.b64encode(v).decode() for k,v in tls.items()}}
    run(['docker','exec','-i',session['worker'],'python3','-c',IMPORT],input=json.dumps(files).encode())
    anchor=run(['docker','container','inspect',session['worker'],'--format','{{.Id}}']).decode().strip()
    phase('issuer_launch')
    args=base_args(session,session['issuer'],'container:'+anchor)+mounts(session,readonly=True)
    # CLI argv carries environment names only. Original credentials stay in the
    # owner subprocess environment and Docker's private container configuration.
    env={'PATH':os.environ.get('PATH',os.defpath),'DOCKER_HOST':'unix:///var/run/docker.sock',**payload['env'],
         'SSL_CERT_FILE':'/recovery-tls/trust.pem'}
    for name in (*payload['env'],'SSL_CERT_FILE'):args+=['--env',name]
    args+=[adapter.IMAGE]
    run(args,env=env)
    phase('issuer_containment')
    adapter.inspect_session(session)
    phase('tls_transport_launch')
    run(['docker','exec','--detach',session['worker'],'python3','-c',TLS])
    phase('complete')


def cleanup_all(session,receipt):
    failed=False
    try:adapter.cleanup({**session,'containers':session['containers']+[session['public_image_probe']]},receipt)
    except Exception:failed=True
    rows=[]
    for name,spec in session['volumes'].items():
        row={'volume':name,'status':'unverified'};rows.append(row)
        try:
            exists=run(['docker','volume','ls','--filter','name=^'+name+'$','--format','{{.Name}}']).decode().splitlines()
            if not exists:
                row['status']='absence_verified'
                continue
            verify_volume(name,spec,session['run_id'])
            run(['docker','volume','rm',name])
            remaining=run(['docker','volume','ls','--filter','name=^'+name+'$','--format','{{.Name}}']).decode().splitlines()
            if remaining:row['status']='still_present';failed=True
            else:row['status']='absence_verified'
        except Exception:failed=True
    receipt['tmpfs_volume_cleanup_obligations']=rows
    receipt['container_cleanup_verified']=not failed
    if failed:raise adapter.Refused()

def register_workdir(work, run_id):
    if work.parent!=STAGING_ROOT or not re.fullmatch('nk49-keycape-'+run_id+'-[a-z0-9_]{8}',work.name):raise adapter.Refused()
    if any(p.is_symlink() for p in (work,*work.parents)):raise adapter.Refused()
    info=work.lstat()
    if not stat.S_ISDIR(info.st_mode) or info.st_uid!=os.getuid() or stat.S_IMODE(info.st_mode)!=0o700:raise adapter.Refused()
    return {'path':str(work),'run_id':run_id,'device':info.st_dev,'inode':info.st_ino,'owner_uid':info.st_uid}


def cleanup_workdir(staging, receipt):
    """Delete only the registered inode below actual /dev/shm, never a replacement."""
    row={'path':staging['path'],'device':staging['device'],'inode':staging['inode'],'status':'unverified'}
    receipt['plaintext_cleanup_obligation']=row
    work=Path(staging['path'])
    if work.parent!=STAGING_ROOT or not re.fullmatch('nk49-keycape-'+staging['run_id']+'-[a-z0-9_]{8}',work.name):raise adapter.Refused()
    if not any(line.split()[1:3]==['/dev/shm','tmpfs'] for line in Path('/proc/mounts').read_text().splitlines()):raise adapter.Refused()
    if any(p.is_symlink() for p in work.parents):raise adapter.Refused()
    parentfd=os.open(work.parent,os.O_RDONLY|os.O_DIRECTORY|os.O_NOFOLLOW)
    try:
        try:info=os.stat(work.name,dir_fd=parentfd,follow_symlinks=False)
        except FileNotFoundError:
            row['status']='absence_verified';receipt['temporary_plaintext_cleanup_verified']=True;return
        if (not stat.S_ISDIR(info.st_mode) or info.st_uid!=staging['owner_uid'] or info.st_uid!=os.getuid()
                or info.st_dev!=staging['device'] or info.st_ino!=staging['inode'] or stat.S_IMODE(info.st_mode)!=0o700):
            row['status']='identity_mismatch_not_removed';raise adapter.Refused()
        def remove(parent,name,expected):
            fd=os.open(name,os.O_RDONLY|os.O_DIRECTORY|os.O_NOFOLLOW,dir_fd=parent)
            try:
                current=os.fstat(fd)
                if (current.st_dev,current.st_ino)!=(expected.st_dev,expected.st_ino):raise adapter.Refused()
                for child in os.listdir(fd):
                    before=os.stat(child,dir_fd=fd,follow_symlinks=False)
                    if before.st_uid!=staging['owner_uid'] or before.st_dev!=staging['device']:raise adapter.Refused()
                    if stat.S_ISDIR(before.st_mode):remove(fd,child,before)
                    elif stat.S_ISREG(before.st_mode) and before.st_nlink==1:os.unlink(child,dir_fd=fd)
                    else:raise adapter.Refused() # symlinks/special files/hardlinks are uncertain, never followed
            finally:os.close(fd)
            after=os.stat(name,dir_fd=parent,follow_symlinks=False)
            if (after.st_dev,after.st_ino)!=(expected.st_dev,expected.st_ino):raise adapter.Refused()
            os.rmdir(name,dir_fd=parent)
        remove(parentfd,work.name,info)
        try:os.stat(work.name,dir_fd=parentfd,follow_symlinks=False)
        except FileNotFoundError:
            row['status']='absence_verified';receipt['temporary_plaintext_cleanup_verified']=True;return
        raise adapter.Refused()
    finally:os.close(parentfd)


def recover_obligations(session, staging, receipt):
    failed=False
    try:cleanup_all(session,receipt)
    except Exception:failed=True
    # The plaintext obligation is attempted even when Docker is unreachable.
    try:cleanup_workdir(staging,receipt)
    except Exception:failed=True
    if failed:raise adapter.Refused()


def process_facts(pid):
    raw=Path('/proc')/str(pid)/'stat'
    try:
        fields=raw.read_text().rsplit(')',1)[1].split()
        return {'pid':pid,'state':fields[0],'pgid':int(fields[2]),'sid':int(fields[3]),
                'start_ticks':int(fields[19]),'uid':raw.parent.stat().st_uid}
    except FileNotFoundError:return None


def isolate_execution():
    # Called by main before reading TLS/key inputs. Never reuse an inherited
    # shell, custody-helper, user or unrelated process group.
    os.setsid()
    pid=os.getpid();facts=process_facts(pid)
    if os.getsid(0)!=pid or os.getpgrp()!=pid or facts['sid']!=pid or facts['pgid']!=pid:raise adapter.Refused()
    return {k:facts[k] for k in ('pid','pgid','sid','start_ticks','uid')}|{'deadline':time.monotonic()+1800}


def verify_execution(execution):
    if execution['pid']!=execution['sid'] or execution['pid']!=execution['pgid'] or execution['uid']!=os.getuid():raise adapter.Refused()
    current=process_facts(execution['pid'])
    if current is not None and any(current[k]!=execution[k] for k in ('pid','pgid','sid','start_ticks','uid')):
        # Numeric leader PID reuse is not authority to signal a new group.
        raise adapter.Refused()


def group_members(execution):
    verify_execution(execution)
    members=[]
    for path in Path('/proc').iterdir():
        if not path.name.isdecimal():continue
        facts=process_facts(int(path.name))
        if facts is None or facts['pgid']!=execution['pgid']:continue
        if facts['sid']!=execution['sid'] or facts['uid']!=execution['uid'] or facts['start_ticks']<execution['start_ticks']:raise adapter.Refused()
        if facts['state'] not in ('Z','X'):members.append(facts)
    return members


def pidfd_dead(pidfd):
    poller=select.poll();poller.register(pidfd,select.POLLIN)
    return bool(poller.poll(0))


def quiesce_execution(execution,parent_pidfd,receipt):
    """Stop only this fresh execution session; cleanup needs no live creator.

    Exact parent pidfd prevents leader reuse. Each remaining session member gets
    its own pidfd after metadata comparison; no inherited/user-group kill occurs.
    Unknown membership or uninterruptible survivors refuse cleanup altogether.
    """
    verify_execution(execution)
    if not pidfd_dead(parent_pidfd):
        signal.pidfd_send_signal(parent_pidfd,signal.SIGTERM)
        poller=select.poll();poller.register(parent_pidfd,select.POLLIN);poller.poll(2000)
        if not pidfd_dead(parent_pidfd):signal.pidfd_send_signal(parent_pidfd,signal.SIGKILL)
    for attempt in range(50):
        members=group_members(execution)
        if pidfd_dead(parent_pidfd) and not members:
            receipt['execution_quiescence_verified']=True;return
        for original in members:
            try:fd=os.pidfd_open(original['pid'])
            except ProcessLookupError:continue
            try:
                current=process_facts(original['pid'])
                if current is None or pidfd_dead(fd):continue
                if current!=original: # state may naturally change; bind stable identity below
                    if any(current[k]!=original[k] for k in ('pid','pgid','sid','start_ticks','uid')):raise adapter.Refused()
                verify_execution(execution)
                signal.pidfd_send_signal(fd,signal.SIGKILL)
            except ProcessLookupError:pass
            finally:os.close(fd)
        time.sleep(.05)
    receipt['execution_quiescence_verified']=False
    raise adapter.Refused()


def daemon_uncertain(staging):
    return any(Path(staging['path']+suffix).exists() or Path(staging['path']+suffix).is_symlink() for suffix in ('-docker-inflight','-docker-uncertain'))


def reaper_main(session,staging,execution,parent_pidfd):
    receipt={'status':'failed','execution_quiescence_verified':False}
    verify_execution(execution)
    if pidfd_dead(parent_pidfd):raise adapter.Refused()
    print('registered',flush=True)
    poller=select.poll();poller.register(parent_pidfd,select.POLLIN)
    remaining=max(0,int((execution['deadline']-time.monotonic())*1000))
    parent_lost=bool(poller.poll(remaining))
    receipt['cleanup_trigger']='parent_exit' if parent_lost else 'execution_deadline'
    try:
        # No deletion races a live creator: first stop/wait this exact owner
        # session and its children. A daemon request already accepted can still
        # complete after CLI loss; retain that uncertainty even after readback.
        quiesce_execution(execution,parent_pidfd,receipt)
        uncertain=daemon_uncertain(staging)
        receipt['in_flight_daemon_request_uncertain']=uncertain
        recover_obligations(session,staging,receipt)
        receipt['status']='bounded_cleanup_uncertain' if uncertain else 'bounded_cleanup_verified'
    except Exception:receipt['status']='bounded_cleanup_uncertain'
    finally:
        os.close(parent_pidfd)
        try:
            fd=os.open(staging['path']+'-reaper.json',os.O_WRONLY|os.O_CREAT|os.O_EXCL|os.O_NOFOLLOW,0o600)
            with os.fdopen(fd,'w') as out:json.dump(receipt,out)
        except Exception:pass


def start_reaper(session, staging, execution):
    adapter.docker_environment()
    verify_execution(execution)
    if session.get('docker_endpoint')!=adapter.ENDPOINT or staging.get('run_id')!=session['run_id']:raise adapter.Refused()
    if register_workdir(Path(staging['path']),session['run_id'])!=staging:raise adapter.Refused()
    parent_pidfd=os.pidfd_open(execution['pid'])
    code="""import importlib.util,json,sys
spec=importlib.util.spec_from_file_location('session',sys.argv[1]);m=importlib.util.module_from_spec(spec);spec.loader.exec_module(m)
m.reaper_main(json.loads(sys.argv[2]),json.loads(sys.argv[3]),json.loads(sys.argv[4]),int(sys.argv[5]))
"""
    try:
        child=subprocess.Popen([sys.executable,'-c',code,str(Path(__file__).resolve()),json.dumps(session),json.dumps(staging),json.dumps(execution),str(parent_pidfd)],
            stdin=subprocess.DEVNULL,stdout=subprocess.PIPE,stderr=subprocess.DEVNULL,
            env={'PATH':os.environ.get('PATH',os.defpath),'DOCKER_HOST':adapter.ENDPOINT},
            close_fds=True,pass_fds=(parent_pidfd,),start_new_session=True)
    finally:os.close(parent_pidfd)
    ready,_,_=select.select([child.stdout],[],[],10)
    if not ready or child.stdout.readline()!=b'registered\n':
        child.terminate();child.wait(timeout=5);raise adapter.Refused()
    child.stdout.close()
    return child


def execute(transfer,admission,tls,receipt,execution,checkpoint=None):
    """Existing private custody/transfer/archive APIs; no second archive format."""
    adapter.manifest_binding_mode(admission,transfer)
    verify_execution(execution)
    if execution['pid']!=os.getpid():raise adapter.Refused()
    target=local.preflight()
    for image in (adapter.IMAGE,):run(['docker','image','inspect',image])
    local.offline.validate(transfer)
    if transfer.get('independent_owner_retrieval_verified') is not True:raise adapter.Refused()
    path=Path(transfer['fetched_path']);bundle.private(path,max_bytes=bundle.LIMIT)
    if (path.stat().st_size!=transfer['ciphertext_bytes'] or local.offline.offsite.digest(path)!=transfer['ciphertext_sha256']
            or admission['ciphertext_sha256']!=transfer['ciphertext_sha256']):raise adapter.Refused()
    if hashlib.sha256(tls['/recovery-tls/trust.pem']).hexdigest()!=admission['trust_sha256']:raise adapter.Refused()
    if hashlib.sha256(tls['/recovery-tls/server.pem']).hexdigest()!=admission['tls_cert_sha256']:raise adapter.Refused()
    # Validate exact private certificate/key pair and original issuer name before
    # opening service containers; do not generate replacement trust or keys.
    from cryptography import x509
    from cryptography.hazmat.primitives import serialization
    cert=x509.load_pem_x509_certificate(tls['/recovery-tls/server.pem'])
    key=serialization.load_pem_private_key(tls['/recovery-tls/server-key.pem'],password=None)
    if cert.public_key().public_numbers()!=key.public_key().public_numbers():raise adapter.Refused()
    keyfd=local.verify.custody.private_key_fd();work=None;session=None;reaper=None;staging=None
    receipt.update(status='failed',target=target,session_source_sha256=hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
                   source_files_sha256={name:hashlib.sha256((HERE/name).read_bytes()).hexdigest() for name in ('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')},
                   temporary_plaintext_cleanup_verified=False,container_cleanup_verified=False,
                   downstream_consumer_reconnect_proven=False,human_mfa_proven=False)
    try:
        if run(['age-keygen','-y',f'/proc/self/fd/{keyfd}'],pass_fds=(keyfd,)).strip()!=bundle.profile()['public_recipient'].encode():raise adapter.Refused()
        session=plan_session()
        work=Path(tempfile.mkdtemp(prefix='nk49-keycape-'+session['run_id']+'-',dir='/dev/shm'))
        staging=register_workdir(work,session['run_id'])
        reaper=start_reaper(session,staging,execution)
        adapter.EXECUTION={**execution,'staging':str(work)}
        receipt.update(execution_binding=execution,bounded_cleanup_reaper_started=True,plaintext_cleanup_obligation={**staging,'status':'registered_before_plaintext'},docker_endpoint=adapter.ENDPOINT,
                       reaper_outcome_path=staging['path']+'-reaper.json',
                       registered_cleanup={'containers':session['containers']+[session['public_image_probe']],'tmpfs_volumes':list(session['volumes'])})
        if checkpoint is not None:checkpoint(receipt)
        archive=work/'archive.tar'
        with archive.open('xb') as output:
            p=adapter.process(['age','-d','-i',f'/proc/self/fd/{keyfd}',str(path)],stdout=output,stderr=subprocess.DEVNULL,pass_fds=(keyfd,),timeout=180)
        if p.returncode or archive.stat().st_size>bundle.LIMIT+4*1024*1024:raise adapter.Refused()
        root=work/'restored';root.mkdir(mode=0o700)
        manifest,_=bundle.restore(archive,root)
        # Data binding comes from exact reviewed ciphertext, never a new grant.
        admission=adapter.private_manifest_admission(admission,manifest)
        payload=adapter.private_payload(root,manifest,admission['manifest_sha256'])
        import yaml
        cfg=yaml.safe_load(base64.b64decode(payload['files']['/etc/keycape/config.yaml']))
        if cfg.get('port')!=8080 or cfg.get('issuer')!=admission['issuer']:raise adapter.Refused()
        # Public executable readback uses the daemon archive API only on the
        # original image rootfs, never a tmpfs or private user-created mount.
        probe=session['public_image_probe']
        try:
            run(['docker','create','--pull=never','--name',probe,'--network','none','--read-only','--label',adapter.LABEL+'='+session['run_id'],adapter.IMAGE])
            binary=adapter.container_file(probe,'/keycape')
            if hashlib.sha256(binary).hexdigest()!=admission['verifier_sha256']:raise adapter.Refused()
        finally:
            receipt['public_image_probe_cleanup_obligations']=[{'container':probe,'status':'unverified'}]
            run(['docker','rm',probe])
            if run(['docker','container','ls','--all','--filter','name=^/'+probe+'$','--format','{{.Names}}']).strip():raise adapter.Refused()
            receipt['public_image_probe_cleanup_obligations'][0]['status']='absence_verified'
        def phase(value):
            # Public enum only: no private payload, exception text or daemon logs.
            if value not in PROVISION_PHASES:raise adapter.Refused()
            receipt['provision_phase']=value
            if checkpoint is not None:checkpoint(receipt)
        provision(session,payload,tls,admission['issuer'],phase)
        receipt.update(ciphertext_sha256=transfer['ciphertext_sha256'],all_manifest_hashes_verified=True,manifest_file_count=len(manifest['files']))
        checked_admission={**admission,'session':session,'trust_path':'/recovery-tls/trust.pem'}
        # Native checker consumes no service secret argument or new grant.
        # Retry only bounded public readiness; no account/factor attempt.
        for attempt in range(40):
            try:
                numbers=serialization.load_pem_private_key(base64.b64decode(payload['files']['/etc/keycape/key.pem']),password=None).public_key().public_numbers()
                adapter.check_jwks(session,checked_admission,numbers);break
            except Exception:
                if attempt==39:raise
                time.sleep(.25)
        adapter.native_checks(root,manifest,checked_admission,receipt,cleanup_session=False)
    finally:
        os.close(keyfd)
        try:
            if session is not None and staging is not None:
                uncertain=daemon_uncertain(staging)
                receipt['in_flight_daemon_request_uncertain']=uncertain
                recover_obligations(session,staging,receipt)
                if uncertain:raise adapter.Refused()
            if reaper is not None:
                reaper.terminate();reaper.wait(timeout=10)
        except Exception:
            receipt['status']='failed'
            raise
        finally:
            adapter.EXECUTION=None
            if work is not None and not work.exists():receipt['temporary_plaintext_cleanup_verified']=True


def persist_receipt(fd,receipt):
    raw=(json.dumps(receipt,indent=2)+'\n').encode()
    os.lseek(fd,0,os.SEEK_SET);os.ftruncate(fd,0)
    with os.fdopen(os.dup(fd),'wb') as output:
        output.write(raw);output.flush();os.fsync(output.fileno())


def main():
    p=argparse.ArgumentParser(description=__doc__)
    for name in ('transfer-receipt','admission','admission-sha256','tls-cert','tls-key','trust','receipt'):p.add_argument('--'+name,required=True)
    args=p.parse_args();os.umask(0o077)
    fd=os.open(args.receipt,os.O_WRONLY|os.O_CREAT|os.O_EXCL|os.O_NOFOLLOW,0o600)
    receipt={'schema':'netkingdom.isolated-keycape-recovery.v1','status':'failed','production_contacted':False,'secret_values_exported':False}
    try:
        execution=isolate_execution()
        raw=Path(args.admission).read_bytes()
        if hashlib.sha256(raw).hexdigest()!=args.admission_sha256:raise adapter.Refused()
        admission=json.loads(raw)
        tls={}
        for name,path in (('server.pem',args.tls_cert),('server-key.pem',args.tls_key),('trust.pem',args.trust)):
            source=Path(path);bundle.private(source,max_bytes=1024*1024);tls['/recovery-tls/'+name]=source.read_bytes()
        execute(json.loads(Path(args.transfer_receipt).read_text()),admission,tls,receipt,execution,checkpoint=lambda r:persist_receipt(fd,r))
    except (Exception,KeyboardInterrupt):receipt.update(status='failed',error='isolated_keycape_failed_private_diagnostics_suppressed')
    finally:
        receipt['finished_at']=datetime.now(timezone.utc).isoformat()
        try:persist_receipt(fd,receipt)
        finally:os.close(fd)
    print('OK: isolated KeyCape native checks and cleanup passed.' if receipt['status']=='isolated_keycape_native_registration_checked' else 'ERROR: isolated KeyCape stopped; review safe receipt.')
    return int(receipt['status']!='isolated_keycape_native_registration_checked')

if __name__=='__main__':
    for sig in (signal.SIGHUP,signal.SIGTERM,signal.SIGINT):signal.signal(sig,local.offline.interrupted)
    raise SystemExit(main())