#!/usr/bin/python3
"""Genuine new-object provisioning in the normal disposable PID1 realm.

No replacement owners, clock grants or launch admission are installed. All
private material is generated in this realm; evidence contains only public
bindings, metadata and authenticated store anchors.
"""
import hashlib
import importlib.util
import json
import os
from pathlib import Path
import pwd
import secrets
import shutil
import socket
import stat
import subprocess
import sys
import time
import traceback

PACKAGE = Path('/opt/c01-fixture')
ROOT = Path('/var/lib/c01-composed')
EVIDENCE = Path('/var/lib/c01-normal')
sys.path.insert(0, str(PACKAGE / 'scripts'))
import c01_access_policy as policy
import c01_access_generation as generation
import c01_access_kernel as kernel
import c01_network as network
import c01_network_owner
import c01_runtime
import c01_nts
import c01_power


def run(*args, check=True, **kw):
    p = subprocess.run(args, capture_output=True, timeout=180, **kw)
    if check and p.returncode:
        raise RuntimeError(f'{args[0]} exit {p.returncode}: {p.stderr.decode(errors="replace")[-2000:]}')
    return p


def write(path, data, mode=0o600):
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open('wb') as stream:
        stream.write(data.encode() if isinstance(data, str) else data)
    path.chmod(mode)


def dump(path, value):
    write(path, json.dumps(value, indent=2, sort_keys=True)+'\n'
          if Path(path).is_relative_to(EVIDENCE) else generation.canonical(value))


def sha(path):
    with Path(path).open('rb') as stream:
        return hashlib.file_digest(stream, 'sha256').hexdigest()


def verify_runtime_dependencies():
    dependencies=json.loads((PACKAGE/'chrony-artifact.json').read_text())
    stack=json.loads((PACKAGE/'service-stack-manifest.json').read_text())['files'] if os.environ.get('C01_RUNNING_SERVICES')=='1' else {}
    for path,item in dependencies.items():
        canonical=path.lstrip('/')
        if canonical.startswith(('lib/','lib64/')):canonical='usr/'+canonical
        assert sha(path)==stack.get(canonical,item)['sha256'], path


def account(name, uid, home='/var/empty', login=False):
    try:
        p = pwd.getpwnam(name)
    except KeyError:
        run('groupadd', '--gid', str(uid), name)
        run('useradd', '--uid', str(uid), '--gid', str(uid), '--no-create-home',
            '--home-dir', home, '--shell', '/bin/bash' if login else '/usr/sbin/nologin', name)
        p = pwd.getpwnam(name)
    assert (p.pw_uid, p.pw_gid, p.pw_dir) == (uid, uid, home)
    return p


def unit(name, body):
    write('/etc/systemd/system/' + name, body, 0o644)


def template(name, replacements):
    text = (PACKAGE / 'image-recipe/configs' / name).read_text()
    for key, value in replacements.items():
        text = text.replace(key, str(value))
    assert '@SUPERVISOR_UNIT@' not in text and '@ORIGINAL_RUNTIME' not in text
    return text


def wait_path(path, seconds=20):
    end = time.monotonic() + seconds
    while time.monotonic() < end:
        if Path(path).exists(): return
        time.sleep(.1)
    raise RuntimeError('owner endpoint absent: ' + str(path))


def structures():
    ROOT.mkdir(mode=0o755)
    work=Path('/var/lib/c01-fixture/home/Work/operator.txt')
    meta=work.stat()
    dump(EVIDENCE/'work-before.json',{'device':meta.st_dev,'inode':meta.st_ino,
        'uid':meta.st_uid,'gid':meta.st_gid,'mode':stat.S_IMODE(meta.st_mode),'sha256':sha(work)})
    # idmapped tmpfs is a real disposable filesystem, not persistent-disk crash
    # qualification. All stores survive process/owner restart within this realm.
    run('mount', '-t', 'tmpfs', '-o', 'mode=0755,size=384m', 'c01-composed', str(ROOT))
    app = account('c01-manager', 61001)
    authority = account('c01-authority', 61002)
    account('archipelago-admin', 1001, '/home/archipelago-admin', True)
    # Real synthetic operator credential, never logged or copied to evidence.
    password = secrets.token_urlsafe(32)
    hashed = run('openssl', 'passwd', '-6', '-stdin', input=password.encode()).stdout.strip()
    run('usermod', '-p', hashed.decode(), 'archipelago-admin')
    write('/run/c01-fixture-admin-password',password)
    admin_home=Path('/home/archipelago-admin')
    assert not list(admin_home.iterdir()), 'fixture must not replace an existing operator home'
    os.chown(admin_home,1001,1001);admin_home.chmod(0o700)
    for role in ('manager', 'snapshot', 'authority'):
        parent = ROOT / ('private-' + role)
        parent.mkdir(mode=0o700)
        backing = parent / 'data'
        if role=='manager' and os.environ.get('C01_LAYOUT_MIGRATION')=='1':
            physical=Path('/var/lib/c01-fixture/data/.c01-private')
            physical.mkdir(mode=0o700)
            (physical/'data').mkdir(mode=0o700)
            (physical/'access').mkdir(mode=0o700)
            policy.STATE.mkdir(parents=True,mode=0o700)
            policy.STATE.chmod(0o700)
            run('mount','--bind',str(physical),str(parent))
            run('mount','--bind',str(physical/'access'),str(policy.STATE))
        else:backing.mkdir(mode=0o700)
        os.chown(backing, 1000, 1000)
    identity = (Path('/var/lib/c01-fixture/data/identity') if os.environ.get('C01_LAYOUT_MIGRATION')=='1'
                else ROOT / 'private-manager/data/identity')
    identity.mkdir(mode=0o700)
    if os.environ.get('C01_LAYOUT_MIGRATION')=='1':identity.chmod(0o755)
    os.chown(identity, 1000, 1000)
    secret = secrets.token_bytes(32)
    der = bytes.fromhex('302e0201010420') + secret + bytes.fromhex('a00706052b8104000a')
    public = run('openssl', 'ec', '-inform', 'DER', '-pubout', '-conv_form', 'compressed',
                 '-outform', 'DER', input=der).stdout[-33:]
    assert public[0] in (2, 3) and len(public) == 33
    write(identity / 'nostr_secret', secret.hex())
    os.chown(identity / 'nostr_secret', 1000, 1000)
    keys = ROOT / 'keys'
    keys.mkdir(mode=0o700)
    for role in ('snapshot', 'authority'):
        write(keys / role, secrets.token_bytes(32))
    proxy = '/usr/local/bin/archipelago-runtime-proxy'
    write(proxy, run('/usr/local/bin/archipelago', 'export-runtime-proxy').stdout, 0o755)
    evidence = {'kernel_release': os.uname().release,
        'clocksource': Path('/sys/devices/system/clocksource/clocksource0/current_clocksource').read_text().strip(),
        'suspend_evidence': None, 'elapsed_error_bound': None,
        'scope': 'actual fixture identity; platform timing qualification unavailable'}
    dump(EVIDENCE / 'platform-input.json', evidence)
    profile = {k: evidence[k] for k in ('kernel_release', 'clocksource')}
    profile.update(evidence_sha256=sha(EVIDENCE / 'platform-input.json'), max_bracket_ns=None,
                   max_observation_error_ns=None, max_elapsed_error_ppb=None, max_awake_ns=None)
    candidate_path = PACKAGE / 'candidate-appointment.json'
    candidate = None
    if candidate_path.exists():
        # This input is privately pinned by the outer qualification manifest.
        # It is not a production-ready flag or a substitute platform identity.
        candidate = json.loads(candidate_path.read_text())
        assert candidate['qualified'] is False and candidate['production_profile_enabled'] is False
        assert candidate['node'] == 'framework' and candidate['max_awake'] == {'secs':120,'nanos':0}
        model = candidate['characterized']
        assert model['identity'] == c01_nts.platform_identity(), 'candidate hardware identity differs'
        assert list(bytes.fromhex(sha(PACKAGE / 'candidate-evidence.json'))) == model['evidence_sha256']
        profile.update(evidence_sha256=sha(PACKAGE / 'candidate-evidence.json'),
                       max_bracket_ns=10_000_000, max_observation_error_ns=250_000_000,
                       max_elapsed_error_ppb=candidate['max_elapsed_error_ppb'],
                       max_awake_ns=120_000_000_000, characterized_model=model)
        dump(EVIDENCE / 'candidate-binding.json', {'candidate_sha256':sha(candidate_path),
             'identity':c01_nts.platform_identity(), 'evidence_sha256':profile['evidence_sha256'],
             'observation_limits':'10ms sampling bracket and 250ms NTS error are refusal limits',
             'production_profile_enabled':False})
    dump('/etc/archipelago-c01/platform.json', profile)
    # Null bounds explicitly fail production PlatformQualification.validate.
    # The byte hash is real; no readiness/suspend evidence is fabricated.
    epoch_backing=ROOT/'private-authority/epoch'
    epoch_backing.mkdir(mode=0o700);os.chown(epoch_backing,1000,1000)
    write(keys/'epoch',secrets.token_bytes(32))
    # This is a non-admitting protocol-test budget, NOT a qualified awake bound.
    # The independent platform profile remains explicitly unqualified above.
    epoch=dict(version=1,backing=str(epoch_backing),key=str(keys/'epoch'),
        delegation=dict(id=list(secrets.token_bytes(32)),allow_wake=True,max_awake={'secs':30,'nanos':0}))
    if candidate is not None:
        epoch['delegation'].update(allow_refresh=True,
                                  max_awake=candidate['max_awake'], characterized=candidate['characterized'])
    dump('/etc/archipelago-c01/epoch.json',epoch)
    dump(EVIDENCE/'epoch-test-scope.json',{'max_awake_seconds':epoch['delegation']['max_awake']['secs'],
         'scope':'private conditional candidate evaluation' if candidate is not None else
                 'non-admitting protocol mechanism; no qualified awake profile'})
    power={'version':1,'operator_uid':1001,'automatic_recovery':'fresh-nts','runtime_retirement':'whole-user-hierarchy',
           'lifecycle_issuer':{'id':secrets.token_hex(32),'scope':'reactivate-current-generation',
             'program_sha256':sha(PACKAGE/'scripts/c01_power_reactivate.py')}}
    dump(c01_power.CONFIG,power)
    c01_power.ROOT.mkdir(mode=0o700)
    c01_power.RUN.mkdir(mode=0o700)
    c01_power.write({'version':1,'sequence':0,'phase':'Open','configuration_sha256':sha(c01_power.CONFIG),
                     'boot_id':Path('/proc/sys/kernel/random/boot_id').read_text().strip(),'nonce':None,
                     'target':None,'coordinator':None,'supervisor':None})
    write(c01_power.ROOT/'lifetime.lock',b'')
    launch = dict(version=4, application_user=app.pw_name, authority_user=authority.pw_name,
        epoch_sha256=sha('/etc/archipelago-c01/epoch.json'),power_sha256=sha(c01_power.CONFIG),
        runtime_uid=1000, runtime_gid=1000, runtime_data_dir='/var/lib/c01-fixture/data',
        **{role+'_backing':str(ROOT / ('private-'+role) / 'data') for role in ('manager','snapshot','authority')},
        snapshot_key=str(keys / 'snapshot'), authority_key=str(keys / 'authority'),
        platform_profile='/etc/archipelago-c01/platform.json', platform_profile_sha256=sha('/etc/archipelago-c01/platform.json'),
        clock_domain=secrets.token_hex(16),
        host_ip=next(info['local'] for interface in json.loads(run('ip','-j','-4','address','show','lo').stdout)
                     for info in interface['addr_info'] if info['scope']=='host'), proxy_path=proxy,
        native_sha256=sha('/usr/local/bin/archipelago'), binding=dict(local_identity=list(public[1:]),
        incarnation=list(secrets.token_bytes(32)), reader_schema=2, writer_schema=2))
    if candidate is not None:
        import c01_admission
        # New disposable identity only. Existing-node identity/asset migration
        # is not performed by this provisioning path.
        required={
            'fips.yaml':('/etc/fips/fips.yaml',b'node:\n  identity:\n    persistent: true\n'),
            'fips.key':('/etc/fips/fips.key',secret.hex().encode()),
            'fips.pub':('/etc/fips/fips.pub',public[1:]),
            'fips.nft':('/etc/fips/fips.nft',b'# Uninstalled FIPS: no kernel rules applied.\n'),
            '80-web-ui.nft':('/etc/fips/fips.d/80-web-ui.nft',b'# Uninstalled FIPS.\n'),
            '85-app-ports.nft':('/etc/fips/fips.d/85-app-ports.nft',b'# No installed apps.\n'),
            '90-ssh.nft':('/etc/fips/fips.d/90-ssh.nft',b'# No FIPS SSH exposure.\n'),
            'wg.public':('/etc/archipelago-fips-wg/wg.public',run('wg','pubkey',input=run('wg','genkey').stdout).stdout),
        }
        for name,(path,data) in required.items():
            assert not Path(path).exists(), 'never overwrite existing identity/assets'
            write(path,data,0o600 if name=='fips.key' else 0o644)
        dump('/etc/archipelago-fips-wg/plan.json',{'version':1,
            'assets':{name:sha(path) for name,(path,_) in required.items()},
            'programs':{'wg':sha('/usr/bin/wg'),'systemctl':sha('/usr/bin/systemctl')},
            'fips':'uninstalled','wg':'uninstalled','peers':[]})
        for path in ('/etc/tor','/var/lib/tor','/var/lib/archipelago-c01/tor',str(Path(launch['runtime_data_dir'])/'tor-hostnames')):
            Path(path).mkdir(parents=True,mode=0o700,exist_ok=True)
        torrc=Path('/etc/tor/torrc')
        if os.environ.get('C01_RUNNING_SERVICES')=='1' and torrc.exists():
            run('systemctl','stop','tor@default.service','tor.service')
            assert torrc.is_file() and not torrc.is_symlink() and torrc.stat().st_uid==0
            prior=torrc.stat();digest=sha(torrc)
            retained=EVIDENCE/'vendor-torrc';assert not retained.exists()
            torrc.rename(retained)
            dump(EVIDENCE/'vendor-torrc-preserved.json',{'inode':prior.st_ino,'sha256':digest,
                 'scope':'new disposable image vendor configuration, not installed-node config'})
        assert not torrc.exists()
        tor_user='User debian-tor\n' if os.environ.get('C01_RUNNING_SERVICES')=='1' else ''
        write('/etc/tor/torrc','# Externally provisioned inactive Tor configuration.\n'+tor_user+'SocksPort 0\nDisableNetwork 1\nDataDirectory /var/lib/tor\n',0o644)
        check=run('/usr/bin/tor','--verify-config','-f','/etc/tor/torrc',check=False)
        dump(EVIDENCE/'tor-config-verification.json',{'exit':check.returncode,'configuration_sha256':sha('/etc/tor/torrc'),
             'service_disposition':'uninstalled','daemon_started':False,
             'stdout':check.stdout.decode(),'stderr':check.stderr.decode()})
        assert check.returncode==0,'genuine Tor config verification refused; see tor-config-verification.json'
        for name in ('filebrowser-credentials.py','indeehub-maintenance-controller.py'):
            write(Path('/opt/archipelago/scripts')/name,(PACKAGE/'scripts'/name).read_bytes(),0o755)
        containment='/opt/archipelago/web-ui/archipelago-runtime/image-recipe/configs/nginx-archipelago.conf'
        write(containment,(PACKAGE/'image-recipe/configs/nginx-archipelago.conf').read_bytes(),0o644)
        contract={'version':1,'platform_sha256':launch['platform_profile_sha256'],
            'required_routes':['management'],
            'roles':sorted(c01_admission.ROLES),'assets':{name:sha(path) for name,path in c01_admission.ASSETS.items()},
            'services':{name:'uninstalled' for name in c01_admission.UNITS},
            'optional_unavailable':sorted(c01_admission.UNITS)}
        raw=(json.dumps(contract,sort_keys=True,indent=2)+'\n').encode()
        digest=hashlib.sha256(raw).hexdigest()
        write(c01_admission.POLICIES/(digest+'.json'),raw,0o600)
        launch.update(version=5,admission_sha256=digest)
        dump(EVIDENCE/'conditional-admission.json',{'policy_sha256':digest,'optional_unavailable':contract['optional_unavailable'],
            'scope':'baseline startup; uninstalled FIPS/WG/Tor/NM features are not claimed accepted'})
    generation.validate_launch(launch)
    dump(policy.LAUNCH, launch)
    runtime = dict(version=1, runtime_user='archipelago', runtime_uid=1000, runtime_gid=1000,
        manager_uid=61001, manager_gid=61001, authority_uid=61002, authority_gid=61002,
        operator_uid=1001, operator_gid=1001, home='/home/archipelago',
        home_backing='/var/lib/c01-fixture/home', runtime_dir='/run/user/1000',
        runtime_data_dir=launch['runtime_data_dir'],
        graphroot='/var/lib/c01-fixture/data/containers/storage' if os.environ.get('C01_HISTORY_MIGRATION')=='1' else '/var/lib/c01-fixture/graphroot',
        subuids=c01_runtime.ranges('/etc/subuid', 'archipelago'), subgids=c01_runtime.ranges('/etc/subgid', 'archipelago'),
        user_cgroup='/user.slice/user-1000.slice/user@1000.service',
        private_parents=[str(ROOT / ('private-'+role)) for role in ('manager','snapshot','authority')])
    dump(policy.RUNTIME, runtime)
    c01_runtime.load_config()
    Path('/var/lib/archipelago-c01').mkdir(mode=0o700, exist_ok=True)
    Path('/run/archipelago-c01').mkdir(mode=0o755, exist_ok=True)
    modules = Path('/usr/lib/archipelago')
    modules.mkdir(exist_ok=True)
    for source in (PACKAGE / 'scripts').glob('*.py'):
        write(modules / source.name, source.read_bytes(), 0o644)
    write(modules / 'c01-provision-entry.py',
          'import json,sys\nsys.path.insert(0,"/usr/lib/archipelago")\nimport c01_provision\n'
          'mode=sys.argv[1]\nprint(json.dumps(c01_provision.run(inspect=mode!="provision",epoch_mode=mode[6:] if mode.startswith("epoch-") else None)),flush=True)\n', 0o644)
    return launch, runtime


def stores():
    receipts = {}
    original=policy.LAUNCH.read_bytes()
    value=json.loads(original)
    marker=Path(value['manager_backing'])/'unreviewed-fixture-history'
    assert not marker.exists()
    write(marker,b'synthetic retained history - must not be discarded\n')
    marker_meta=marker.stat();marker_digest=sha(marker)
    rejected=run('systemd-run','--wait','--pipe','--unit=c01-store-existing-manager',
        '-p','Delegate=yes','-p','DelegateSubgroup=supervisor','-p','NoNewPrivileges=yes',
        '-p','NetworkNamespacePath=/run/netns/archipelago-managed',
        '/usr/bin/python3','-B','/usr/lib/archipelago/c01-provision-entry.py','provision',check=False)
    assert rejected.returncode!=0 and b'existing manager state requires' in rejected.stderr
    assert marker.stat().st_ino==marker_meta.st_ino and sha(marker)==marker_digest
    assert all(not list(Path(value[role+'_backing']).iterdir()) for role in ('snapshot','authority'))
    dump(EVIDENCE/'existing-manager-refusal.json',{'exit':rejected.returncode,'stderr':rejected.stderr.decode(),
        'marker_inode':marker_meta.st_ino,'marker_sha256':marker_digest,'stores_remained_empty':True})
    marker.unlink() # teardown only this exact marker created by this fixture
    before=None
    for label,operation in [('provision','provision'),('inspect','inspect'),('reprovision','provision'),
                            ('wrong-key','inspect'),('wrong-binding','inspect')]:
        if label=='wrong-key':
            changed=dict(value,snapshot_key=str(ROOT/'keys/wrong-snapshot'))
            write(changed['snapshot_key'],secrets.token_bytes(32))
            dump(policy.LAUNCH,changed)
        elif label=='wrong-binding':
            changed=dict(value,binding={**value['binding'],'incarnation':list(secrets.token_bytes(32))})
            dump(policy.LAUNCH,changed)
        else:
            write(policy.LAUNCH,original)
        index = len(receipts)
        p = run('systemd-run', '--wait', '--pipe', '--unit=c01-store-' + str(index),
            '-p', 'Delegate=yes', '-p', 'DelegateSubgroup=supervisor', '-p', 'NoNewPrivileges=yes',
            '-p', 'NetworkNamespacePath=/run/netns/archipelago-managed',
            '/usr/bin/python3', '-B', '/usr/lib/archipelago/c01-provision-entry.py', operation, check=False)
        receipts[str(index)] = {'case':label, 'operation':operation, 'exit':p.returncode,
            'stdout':p.stdout.decode(), 'stderr':p.stderr.decode()}
        write(policy.LAUNCH,original)
        inventory={str(path.relative_to(ROOT)): {'sha256':sha(path), 'uid':path.stat().st_uid,
                     'gid':path.stat().st_gid,'inode':path.stat().st_ino}
            for role in ('snapshot','authority') for path in Path(value[role+'_backing']).iterdir() if path.is_file()}
        if before is None: before=inventory
        else: assert inventory==before, 'initial stores changed during inspection/refusal'
        assert all(item['uid']==item['gid']==1000 for item in inventory.values()), 'idmapped storage changed original ownership'
        dump(EVIDENCE/'store-preservation.json',inventory)
        dump(EVIDENCE / 'stores.json', receipts)
        assert (p.returncode == 0) == (index < 2), receipts[str(index)]
    assert json.loads(receipts['0']['stdout'])['anchor'] == json.loads(receipts['1']['stdout'])['anchor']
    return receipts


def network_owner():
    if c01_network_owner.CONFIG.exists():
        current=c01_network_owner.Policy()
        try:
            assert current.value['version']==2
            current.check()
        finally:current.close()
        wait_path(c01_network_owner.CONTROL)
        return {'policy_sha256':sha(c01_network_owner.CONFIG),'retained_pre_runtime_owner':True}
    import runtime_uplink
    value = dict(version=1, namespaces={'host':list(network.process_namespace(1)),
        'managed':list(network.appointed('managed')), 'runtime':list(network.appointed('runtime'))},
        manifests=[], routes=[dict(name='management',source='host',target='managed',protocol='tcp',port=5678,
        app_id=None,projection_sha256=None,target_cgroup='/system.slice/archipelago.service/c01-application',target_uid=61001)])
    value.update(version=2,uplink={'executable':'/usr/bin/pasta.avx2','sha256':sha('/usr/bin/pasta.avx2'),
                                 'wan':list(runtime_uplink.ready()),'dns':['1.1.1.1','1.0.0.1']})
    dump(c01_network_owner.CONFIG, value)
    unit('archipelago-c01-network.service', template('archipelago-c01-network.service', {}))
    write('/etc/systemd/system/user@1000.service.d/82-c01-runtime-uplink-owner.conf',
          template('82-c01-runtime-uplink-owner.conf',{}),0o644)
    write('/etc/systemd/system/user-runtime-dir@1000.service.d/83-c01-runtime-directory-lifetime.conf',
          template('83-c01-runtime-directory-lifetime.conf.in',{'@ORIGINAL_RUNTIME_UID@':'1000'}),0o644)
    run('systemctl','daemon-reload')
    run('systemctl','start','archipelago-c01-network.service')
    wait_path(c01_network_owner.CONTROL)
    return {'policy_sha256':sha(c01_network_owner.CONFIG), 'main_pid':run('systemctl','show',
        'archipelago-c01-network.service','-p','MainPID','--value').stdout.decode().strip()}


def epoch_owner():
    results=[]
    for mode in ('provision','recover'):
        p=run('systemd-run','--wait','--pipe','--unit=c01-epoch-'+mode,
            '-p','Delegate=yes','-p','DelegateSubgroup=supervisor','-p','NoNewPrivileges=yes',
            '-p','NetworkNamespacePath=/run/netns/archipelago-managed',
            '/usr/bin/python3','-B','/usr/lib/archipelago/c01-provision-entry.py','epoch-'+mode)
        result=json.loads(p.stdout);results.append(result)
        assert result['clock_authorized'] is False and result['revoked_aliases'] is True
    first,second=(item['epoch_context'] for item in results)
    assert first['cause']=='Restart' and second['cause']=='Wake' and second['epoch']>first['epoch']
    assert second['not_before']!=first['not_before'] and second['application_owner']!=first['application_owner']
    assert second['nonce']!=first['nonce'] and second['domain']!=first['domain']
    legacy=run('systemd-run','--wait','--pipe','--unit=c01-epoch-legacy-refusal',
        '-p','Delegate=yes','-p','DelegateSubgroup=supervisor','-p','NoNewPrivileges=yes',
        '-p','NetworkNamespacePath=/run/netns/archipelago-managed',
        '/usr/bin/python3','-B','/usr/lib/archipelago/c01-provision-entry.py','inspect',check=False)
    assert legacy.returncode!=0, 'epoch-enabled anchor silently downgraded'
    dump(EVIDENCE/'native-epoch.json',{'lifetimes':results,'legacy_exit':legacy.returncode,'legacy_stderr':legacy.stderr.decode()})
    return {'epoch':second['epoch'],'cause':second['cause'],'clock_authorized':False}


def d09_module():
    spec = importlib.util.spec_from_file_location('compose_d09', PACKAGE / 'scripts/provision-d09-lifecycle.py')
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def prepare_owners():
    launch, _ = policy.json_read(policy.LAUNCH)
    runtime, _ = policy.json_read(policy.RUNTIME)
    if launch['version']==5:
        credentials=json.loads((PACKAGE/'baseline-auth.json').read_text())
        destination=Path(launch['runtime_data_dir'] if os.environ.get('C01_LAYOUT_MIGRATION')=='1'
                         else launch['manager_backing'])/'user.json'
        assert not destination.exists(), 'baseline credential provisioning cannot overwrite a user'
        dump(destination,credentials['user']);os.chown(destination,1000,1000);destination.chmod(0o600)
        history=Path(runtime['home_backing'])/'.bash_history'
        assert not history.exists()
        write(history,'# retained qualification history\n',0o600);os.chown(history,1000,1000)
        write('/etc/systemd/system/archipelago.service.d/45-retained-operator.conf',
              '# Retained operator fragment: no startup override.\n',0o644)
    unit('archipelago-c01-power-barrier@.service',c01_power.barrier_unit())
    unit('archipelago-c01-power-observer.service',c01_power.observer_unit())
    unit('archipelago-c01-access-requalify.service',c01_power.access_requalify_unit())
    unit('archipelago-c01-power-boot.service',c01_power.boot_unit())
    write('/usr/libexec/archipelago-c01-reactivate',(PACKAGE/'scripts/c01_power_reactivate.py').read_bytes(),0o755)
    for target in c01_power.TARGETS:
        write(Path('/etc/systemd/system')/(target+'.service.d/40-c01-power.conf'),c01_power.power_dropin(target),0o644)
    repairs=[]
    for name in ('archipelago-doctor.service','archipelago-doctor.timer','archipelago-fips.service',
                 'archipelago-tor-helper.path','archipelago-tor-helper.service','archipelago-update.service',
                 'archipelago-update.timer','archipelago-wg-address.service','archipelago-wg.service','nostr-relay.service'):
        path=Path('/etc/systemd/system')/name
        if not path.exists() and not path.is_symlink():
            repairs.append({'path':str(path),'status':'absent from minimal rootfs'})
            continue
        meta=path.lstat()
        assert path.is_file() and not path.is_symlink() and meta.st_uid==meta.st_gid==0 and meta.st_nlink==1
        assert meta.st_mode & 0o7777 in (0o644,0o664)
        if meta.st_mode & 0o7777 == 0o664:
            digest=sha(path)
            path.chmod(0o644)
            assert path.stat().st_ino==meta.st_ino and sha(path)==digest
            repairs.append({'path':str(path),'before_mode':'0664','after_mode':'0644','sha256':digest,'inode':meta.st_ino})
    dump(EVIDENCE/'image-unit-protection.json',repairs)
    # Keep the production no-directory-alias inventory rule. Normalize only the
    # image's known usr-merge systemd links, proving the exact target inode and
    # bytes before/after; this does not change enable/mask/stop decisions.
    lib=Path('/lib')
    assert lib.is_symlink() and lib.lstat().st_uid==0 and os.readlink(lib) in ('usr/lib','/usr/lib')
    links=[]
    for path in Path('/etc/systemd/system').rglob('*'):
        if not path.is_symlink(): continue
        target=os.readlink(path)
        if not target.startswith('/lib/systemd/system/'): continue
        assert path.lstat().st_uid==0
        canonical='/usr'+target
        policy.protected(Path(canonical))
        before=path.stat(); digest=sha(path)
        assert Path(canonical).stat().st_ino==before.st_ino and sha(canonical)==digest
        path.unlink();path.symlink_to(canonical)
        assert path.stat().st_ino==before.st_ino and sha(path)==digest
        links.append({'path':str(path),'before':target,'after':canonical,'sha256':digest,'inode':before.st_ino})
    dump(EVIDENCE/'image-unit-links.json',links)
    # The image's legacy root doctor/update/Tor helpers can otherwise recreate
    # UID1000 callers during the real transition. Retire these exact activation
    # routes as external fixture deployment, retaining their original inodes.
    legacy_units=('archipelago-doctor.timer','archipelago-doctor.service',
                  'archipelago-update.timer','archipelago-update.service',
                  'archipelago-tor-helper.path','archipelago-tor-helper.service')
    retired=[]
    archive=Path('/var/lib/archipelago-c01/legacy-units')
    archive.mkdir(mode=0o700,exist_ok=True)
    present_legacy=[name for name in legacy_units if (Path('/etc/systemd/system')/name).exists() or
                    (Path('/etc/systemd/system')/name).is_symlink()]
    if present_legacy: run('systemctl','stop',*present_legacy)
    for name in legacy_units:
        path=Path('/etc/systemd/system')/name
        if name not in present_legacy:
            retired.append({'unit':name,'status':'absent; no retirement claimed'})
            continue
        meta=path.lstat();digest=sha(path)
        assert path.is_file() and not path.is_symlink() and meta.st_uid==0 and meta.st_nlink==1
        destination=archive/name
        assert not destination.exists()
        path.rename(destination);path.symlink_to('/dev/null')
        assert destination.stat().st_ino==meta.st_ino and sha(destination)==digest
        retired.append({'unit':name,'inode':meta.st_ino,'sha256':digest,'retained':str(destination)})
    run('systemctl','daemon-reload')
    dump(EVIDENCE/'legacy-root-activation-retirement.json',retired)
    run('systemctl', 'stop', 'user@1000.service')
    assert all(1000 not in item['uids'] for item in kernel.all_processes()), 'fixture runtime writer not retired'
    # External deployment custody of immutable public payloads, not runtime
    # wallet/data ownership repair. Keep every byte/inode and refuse aliases.
    custody=[]
    web=Path('/opt/archipelago/web-ui')
    if not web.exists():
        custody.append({'path':str(web),'status':'absent; asset qualification remains open'})
    for path in [p for p in [web.parent, web, *sorted(web.rglob('*'))] if p.exists()]:
        meta=path.lstat()
        assert meta.st_uid in (0,1000) and meta.st_gid in (0,1000)
        assert stat.S_ISDIR(meta.st_mode) or stat.S_ISREG(meta.st_mode) and meta.st_nlink==1
        digest=sha(path) if path.is_file() else None
        mode=stat.S_IMODE(meta.st_mode)&~0o022
        os.chown(path,0,0);path.chmod(mode)
        assert path.stat().st_ino==meta.st_ino and (not path.is_file() or sha(path)==digest)
        if meta.st_uid or meta.st_gid or stat.S_IMODE(meta.st_mode)!=mode:
            custody.append({'path':str(path),'before':[meta.st_uid,meta.st_gid,stat.S_IMODE(meta.st_mode)],
                            'after':[0,0,mode],'inode':meta.st_ino,'sha256':digest})
    dump(EVIDENCE/'image-public-payload-custody.json',custody)
    containment=web/'archipelago-runtime/image-recipe/configs/nginx-archipelago.conf'
    compiled_template=(PACKAGE/'image-recipe/configs/nginx-archipelago.conf').read_bytes()
    if containment.exists():
        assert containment.read_bytes()==compiled_template, 'existing containment template requires explicit disposition'
    else:
        write(containment,compiled_template,0o644)
    dump(EVIDENCE/'image-containment-template.json',{'path':str(containment),'sha256':sha(containment),
        'source':'frozen image-recipe/configs/nginx-archipelago.conf'})
    home = Path(runtime['home_backing'])
    (home / '.config/systemd/user/default.target.wants/c01-fixture-runtime.service').unlink()
    for name in ('archipelago-runtime-worker.service','archipelago-runtime-podman.service'):
        write(home / '.config/systemd/user' / name, template(name, {}), 0o644)
    (home / '.config/systemd/user/default.target.wants/archipelago-runtime-worker.service').symlink_to('../archipelago-runtime-worker.service')
    replacements = {'@ORIGINAL_RUNTIME_UID@':'1000','@ORIGINAL_HOME@':'/home/archipelago',
        '@SANITIZED_RUNTIME_HOME@':runtime['home_backing'],
        **{'@'+role.upper()+'_BACKING_PARENT@':str(Path(launch[role+'_backing']).parent) for role in ('manager','snapshot','authority')}}
    write('/etc/systemd/system/user@1000.service.d/80-archipelago-runtime-user.conf',
        template('80-archipelago-runtime-user.conf.in', replacements), 0o644)
    write('/etc/systemd/system/user@1000.service.d/81-c01-runtime-access.conf',
        template('81-c01-runtime-access.conf.in', {'@'+role.upper()+'_PARENT@':str(Path(launch[role+'_backing']).parent)
            for role in ('manager','snapshot','authority')}), 0o644)
    unit('archipelago-c01-access.service', template('archipelago-c01-access.service.in',
        {'@SUPERVISOR_UNIT@':'archipelago.service','@ORIGINAL_RUNTIME_DATA@':runtime['runtime_data_dir']}))
    unit('archipelago-runtime-broker.service', template('archipelago-runtime-broker.service.in',
        {'@RUNTIME_DATA_DIR@':runtime['runtime_data_dir'],'@ORIGINAL_RUNTIME_UID@':'1000'}))
    unit('archipelago.service', '[Service]\nType=notify\nNotifyAccess=main\nDelegate=yes\n'
        'DelegateSubgroup=supervisor\nNoNewPrivileges=yes\nUser=root\nGroup=root\n'
        'NetworkNamespacePath=/run/netns/archipelago-managed\n'
        'ExecStart=/usr/local/bin/archipelago c01-supervisor-v3\n'
        'TimeoutStartSec=150\nTimeoutStopSec=infinity\nRestart=no\n')
    write('/etc/systemd/system/archipelago.service.d/76-c01-power.conf',c01_power.supervisor_dropin(),0o644)
    if (PACKAGE/'nts-addresses.json').exists():
        verify_runtime_dependencies()
        account('archipelago-nts',61003,'/nonexistent')
        addresses=json.loads((PACKAGE/'nts-addresses.json').read_text())
        with Path('/etc/hosts').open('a') as stream:
            for name,values in sorted(addresses.items()):
                assert name in ('time.cloudflare.com','nts.netnod.se')
                for address in values: stream.write(f'\n{address} {name}\n')
        write(c01_nts.CONFIG,c01_nts.configuration(c01_nts.Policy(frozenset(addresses))),0o644)
        unit('archipelago-nts.service',template('archipelago-nts.service',{}))
        for path,mode in [('/var/lib/archipelago-network-settings',0o700),('/etc/archipelago',0o755),('/etc/fips/fips.d',0o755),('/etc/tor',0o755)]:
            Path(path).mkdir(parents=True,mode=mode,exist_ok=True)
        write('/etc/systemd/system/archipelago.service.d/70-c01-supervisor.conf',
            template('70-c01-supervisor.conf.in',{'@PRIVATE_BACKING_PARENT@':' '.join(runtime['private_parents']),
                     '@RUNTIME_DATA_ROOT@':runtime['runtime_data_dir']}),0o644)
    run('systemctl', 'daemon-reload')
    provision=d09_module()
    provision.install_package()
    if os.environ.get('C01_POWER_COMPONENT') == '1':
        # Explicit data-only native lifetime. It cannot authorize time, start
        # the app or emit APP_READY. Do not submit a D09 start grant for it.
        write('/etc/systemd/system/archipelago.service.d/75-c01-epoch-component.conf',
            '[Service]\nType=notify\nExecStart=\n'
            'ExecStart=/usr/local/bin/archipelago c01-epoch-lifetime-v1\n',0o644)
        run('systemctl','daemon-reload')
    # Qualify the candidate's actual systemd projection before any access or
    # D09 lease exists. No unit is started. Restore the exact absent preimage
    # before inventory/transition; never modify config behind a live lease.
    workspace=provision.owner.DESTINATIONS['config']
    assert not workspace.exists() and not workspace.is_symlink()
    old_effective=provision.owner.effective_configuration()
    workspace_bytes=Path('/usr/libexec/archipelago-d09-terminal-workspace.conf').read_bytes()
    write(workspace,workspace_bytes)
    run('systemctl','daemon-reload')
    candidate_effective=provision.owner.effective_configuration()
    assert workspace.read_bytes()==workspace_bytes
    workspace.unlink()
    run('systemctl','daemon-reload')
    assert provision.owner.effective_configuration()==old_effective
    dump(EVIDENCE/'candidate-systemd-config.json', {'old':old_effective,'new':candidate_effective,
        'old_sha256':hashlib.sha256(old_effective.encode()).hexdigest(),
        'new_sha256':hashlib.sha256(candidate_effective.encode()).hexdigest(),
        'started':False,'preimage_restored':True})
    runtime['version'] = 2
    runtime['workspace_helper_sha256'] = sha('/usr/libexec/archipelago-d09-prepare-workspace')
    dump(policy.RUNTIME, runtime)
    # Production SSH/polkit services, not successful-command stubs.
    Path('/run/sshd').mkdir(mode=0o755, exist_ok=True)
    run('ssh-keygen','-q','-t','ed25519','-N','','-f','/etc/ssh/c01-fixture-host')
    run('ssh-keygen','-q','-t','ed25519','-N','','-f','/run/c01-fixture-admin-key')
    write('/etc/ssh/c01-authorized/archipelago-admin',Path('/run/c01-fixture-admin-key.pub').read_bytes(),0o644)
    write('/run/c01-fixture-known-hosts',b'[127.0.0.1]:22222 '+Path('/etc/ssh/c01-fixture-host.pub').read_bytes())
    write('/etc/ssh/sshd_config', 'HostKey /etc/ssh/c01-fixture-host\nUsePAM no\nPort 22222\n'
        'ListenAddress 127.0.0.1\nAuthorizedKeysFile /etc/ssh/c01-authorized/%u\n'
        'PasswordAuthentication no\nKbdInteractiveAuthentication no\n', 0o644)
    run('systemctl','start','ssh.service','polkit.service')
    operator_route(False)
    if (PACKAGE/'nts-addresses.json').exists(): run('systemctl','start','archipelago-nts.service')
    return {'runtime_sha256':sha(policy.RUNTIME), 'd09_modules_installed':True}


def conditional_contract():
    import c01_admission
    launch,_=policy.json_read(policy.LAUNCH)
    require_candidate=launch['version']==5
    if not require_candidate:return {'scope':'legacy non-admitting composition'}
    contract=c01_admission.Contract(launch)
    try:
        contract.check()
        return {'measured_assets':len(contract.assets),'services':contract.initial,
                'policy_sha256':launch['admission_sha256']}
    finally:contract.close()


def role_view_probe():
    launch,_=policy.json_read(policy.LAUNCH)
    parents=' '.join(str(Path(launch[name+'_backing']).parent) for name in ('manager','snapshot','authority'))
    result=run('systemd-run','--wait','--pipe','--unit=c01-role-view-probe',
        '-p','Delegate=yes','-p','DelegateSubgroup=supervisor',
        '-p','User=root','-p','Group=root','-p','AmbientCapabilities=CAP_SETUID',
        '-p','UMask=0077',
        '-p','NoNewPrivileges=yes','-p','PrivateTmp=yes','-p','PrivateDevices=yes',
        '-p','ProtectHome=yes','-p','ProtectSystem=strict',
        '-p','ReadWritePaths=/sys/fs/cgroup /var/lib/archipelago-c01 '+parents+' '+launch['runtime_data_dir'],
        '-p','NetworkNamespacePath=/run/netns/archipelago-managed',
        '/usr/bin/python3','-B',str(PACKAGE/'tests/regression/c01-role-views.py'),check=False)
    write(EVIDENCE/'role-view-probe.log',result.stdout+result.stderr)
    assert result.returncode==0,'actual authority view sandbox refused; see role-view-probe.log'
    return {'actual_idmapped_views':True}


def operator_route(after):
    ssh=['/usr/bin/ssh','-F','/dev/null','-p','22222','-i','/run/c01-fixture-admin-key',
         '-o','BatchMode=yes','-o','IdentitiesOnly=yes','-o','StrictHostKeyChecking=yes',
         '-o','UserKnownHostsFile=/run/c01-fixture-known-hosts','archipelago-admin@127.0.0.1']
    assert run(*ssh,'id -u').stdout.strip()==b'1001', 'actual operator SSH route failed'
    result={'ssh_uid':1001,'host_key_sha256':sha('/etc/ssh/c01-fixture-host.pub')}
    if after:
        password=Path('/run/c01-fixture-admin-password').read_bytes()+b'\n'
        assert run(*ssh,"sudo -S -p '' id -u",input=password).stdout.strip()==b'0', 'actual operator sudo route failed'
        result['sudo_uid']=0
    dump(EVIDENCE/('operator-after.json' if after else 'operator-before.json'),result)


def nts_prepare():
    """Start the independent observer even when another owner cannot provision."""
    assert (PACKAGE/'nts-addresses.json').exists(), 'NTS dependencies absent'
    verify_runtime_dependencies()
    account('archipelago-nts',61003,'/nonexistent')
    addresses=json.loads((PACKAGE/'nts-addresses.json').read_text())
    with Path('/etc/hosts').open('a') as stream:
        for name,values in sorted(addresses.items()):
            assert name in ('time.cloudflare.com','nts.netnod.se')
            for address in values: stream.write(f'\n{address} {name}\n')
    write(c01_nts.CONFIG,c01_nts.configuration(c01_nts.Policy(frozenset(addresses))),0o644)
    unit('archipelago-nts.service',template('archipelago-nts.service',{}))
    run('systemctl','daemon-reload')
    run('systemctl','start','archipelago-nts.service')
    return {'configuration_sha256':sha(c01_nts.CONFIG),'clock_setting':False}


def nts_owner():
    assert (PACKAGE/'nts-addresses.json').exists(), 'fresh same-realm NTS dependencies/egress not supplied'
    failures=[]
    started=time.monotonic()
    end=started+300
    while time.monotonic()<end:
        try:
            observed=subprocess.run(['/usr/local/bin/archipelago','c01-observe-nts-v1'],
                capture_output=True, timeout=max(.001,end-time.monotonic()))
        except subprocess.TimeoutExpired:
            failures.append('native observation exceeded independent 300s prewarm deadline')
            dump(EVIDENCE/'same-realm-nts-refusals.json',failures)
            break
        if observed.returncode==0:
            value=json.loads(observed.stdout)
            pid=int(run('systemctl','show','archipelago-nts.service','-p','MainPID','--value').stdout)
            status=dict(line.split(':',1) for line in Path(f'/proc/{pid}/status').read_text().splitlines() if ':' in line)
            assert status['NoNewPrivs'].strip()=='1'
            assert all(int(status[k],16)&(1<<25)==0 for k in ('CapEff','CapPrm','CapAmb'))
            result={'observation':value,'observer_pid':pid,'native_sha256':sha('/usr/local/bin/archipelago'),
                    'prewarm_elapsed_seconds':time.monotonic()-started,'prewarm_budget_seconds':300,
                    'scope':'readiness only; each epoch still requires a fresh post-barrier reference',
                    'recovery_grant_issued':False,'platform_suspend_qualified':False,'prior_refusals':failures,
                    'egress':run('/usr/sbin/nft','list','ruleset').stdout.decode()}
            dump(EVIDENCE/'same-realm-nts.json',result)
            probes=[]
            for writable in (False,True):
                args=['systemd-run','--wait','--pipe','--unit=c01-nts-sandbox-'+str(int(writable)),
                      '-p','NoNewPrivileges=yes','-p','PrivateDevices=yes','-p','ProtectHome=yes',
                      '-p','ProtectSystem=strict','-p','PrivateTmp=yes']
                if writable:args+=['-p','ReadWritePaths=/run/archipelago-nts']
                check=run(*args,'/usr/bin/chronyc','-n','-c','-h',str(c01_nts.SOCKET),'tracking',check=False)
                probes.append({'writable_command_directory':writable,'exit':check.returncode,
                               'stdout':check.stdout.decode(),'stderr':check.stderr.decode()})
            dump(EVIDENCE/'nts-command-sandbox.json',probes)
            assert probes[-1]['exit']==0,'real NTS command sandbox unavailable'
            return result
        failures.append(observed.stderr.decode()[-2000:])
        dump(EVIDENCE/'same-realm-nts-refusals.json',failures)
        time.sleep(min(4,max(0,end-time.monotonic())))
    raise ValueError('fresh same-realm native NTS prewarm unavailable after300s')


def access_owner(post_retirement=None,migration_layout=None):
    launch, _ = policy.json_read(policy.LAUNCH)
    runtime, _ = policy.json_read(policy.RUNTIME)
    for path in (policy.STAGED, policy.STATE, policy.STATE/'quarantine'):
        path.mkdir(parents=True, mode=0o700, exist_ok=True)
        path.chmod(0o700)
    for path in ('/etc/polkit-1/rules.d','/etc/cron.d','/var/spool/cron'):
        Path(path).mkdir(parents=True, mode=0o755, exist_ok=True)
    # A real known legacy copy + live process holding those bytes. Both are
    # enumerated, frozen and retired by production transition, never attested by
    # an invented journal. The retained original remains behind private ancestry.
    legacy = Path('/var/lib/c01-legacy-secret')
    identity_source=Path(migration_layout['source_root'] if migration_layout is not None else launch['manager_backing'])/'identity/nostr_secret'
    write(legacy, identity_source.read_bytes())
    os.chown(legacy, 1000, 1000)
    Path('/run/c01-legacy').mkdir(mode=0o700)
    os.chown('/run/c01-legacy',1000,1000)
    write('/usr/lib/archipelago/c01-legacy-fixture.py',
        'import os,pathlib,time\nkey=pathlib.Path("/var/lib/c01-legacy-secret").read_bytes()\n'
        'assert len(key)==64\npathlib.Path("/run/c01-legacy/ready").write_text(str(os.getpid()))\n'
        'while True:time.sleep(1)\n', 0o644)
    run('systemd-run','--unit=c01-legacy','-p','User=archipelago','-p','Restart=no',
        '/usr/bin/python3','-B','/usr/lib/archipelago/c01-legacy-fixture.py')
    wait_path('/run/c01-legacy/ready')
    group = Path('/sys/fs/cgroup/system.slice/c01-legacy.service')
    members = kernel.members(group)
    assert members
    assert int(Path('/run/c01-legacy/ready').read_text()) in members
    names = ['archipelago','c01-manager','c01-authority']
    before = {name:path.read_bytes() if path.exists() else None for name,path in policy.TARGETS.items()}
    future = {name:raw for name,raw in before.items() if raw is not None}
    for name in ('passwd','shadow','group','gshadow'):
        rows = [line.split(':') for line in before[name].decode().splitlines()]
        for row in rows:
            if name=='passwd' and row[0] in names: row[6]='/usr/sbin/nologin'
            if name=='shadow' and row[0] in names and not row[1].startswith('!'): row[1]='!'+row[1]
            if name in ('group','gshadow'):
                for field in ((3,) if name=='group' else (2,3)):
                    row[field]=','.join(x for x in row[field].split(',') if x not in names)
        future[name]=('\n'.join(':'.join(row) for row in rows)+'\n').encode()
    future['sudoers']=policy.sudo_policy('archipelago-admin')
    future['polkit']=policy.polkit_policy(names)
    future['sshd']=policy.ssh_policy(before['sshd'],names)
    future['cron-deny']=future['at-deny']=('\n'.join(names)+'\n').encode()
    for name, raw in future.items(): write(policy.STAGED/name, raw)
    masks=['c01-legacy.service','user@61001.service','user@61002.service']
    # Compute exact postimage inventory, then restore every preimage BEFORE the
    # real owner transaction. Version2 excludes only appointed D09 workspace.
    for name, raw in future.items():
        write(policy.TARGETS[name],raw,0o600 if name in ('shadow','gshadow') else 0o440 if name=='sudoers' else 0o644)
    for name in masks: Path('/etc/systemd/system',name).symlink_to('/dev/null')
    trees={path:policy.tree_digest(Path(path),(generation.WORKSPACE,generation.STAGING)) for path in (
        '/etc/sudo.conf','/etc/nsswitch.conf','/etc/pam.d','/etc/ssh','/etc/polkit-1',
        '/usr/share/polkit-1/rules.d','/etc/systemd/system','/etc/cron.d','/var/spool/cron','/etc/subuid','/etc/subgid')}
    if post_retirement is not None:
        trees['/usr/lib/archipelago']=policy.tree_digest(Path('/usr/lib/archipelago'))
    for name, raw in before.items():
        if raw is None: policy.TARGETS[name].unlink()
        else: write(policy.TARGETS[name],raw,0o600 if name in ('shadow','gshadow') else 0o440 if name=='sudoers' else 0o644)
    for name in masks: Path('/etc/systemd/system',name).unlink()
    source=Path(launch['manager_backing'])/'identity/nostr_secret'
    meta=group.stat()
    node_appointment=json.loads((PACKAGE/'node-cutover-fixture.json').read_text()) if (PACKAGE/'node-cutover-fixture.json').exists() else None
    migration_authority='one-qualified-guarded-yaya-migration' if node_appointment and node_appointment['node']=='yaya' else 'one-qualified-guarded-framework-migration'
    value=dict(version=2, authorization=migration_authority,
        inventory_scope='complete-known-writers-and-copies-no-unresolved-signers',
        runtime_sha256=sha(policy.RUNTIME),launch_sha256=sha(policy.LAUNCH),
        operator=dict(name='archipelago-admin',uid=1001,gid=1001),
        accounts={role:dict(name=name,uid=uid,gid=uid) for role,name,uid in
            [('runtime','archipelago',1000),('manager','c01-manager',61001),('authority','c01-authority',61002)]},
        files={name:dict(before=hashlib.sha256(before[name]).hexdigest() if before[name] is not None else None,
                        after=hashlib.sha256(raw).hexdigest()) for name,raw in future.items()},trees=trees,
        cohorts=[dict(path=str(group),device=meta.st_dev,inode=meta.st_ino,
                      members={str(pid):kernel.process(pid)['start'] for pid in members})],
        copies=[dict(source=str(legacy),identity=policy.identity(legacy.stat()),disposition='quarantine-known-copy'),
                dict(source=str(source),identity=policy.identity(identity_source.stat()),disposition='retained-exclusive-backing')],
        preserved=[dict(path=runtime[k],identity=policy.identity(Path(runtime[k]).stat())) for k in ('home','graphroot','runtime_data_dir')],
        restart_units=masks,broker_cgroup='/sys/fs/cgroup/system.slice/archipelago-runtime-broker.service',
        supervisor_cgroup='/sys/fs/cgroup/system.slice/archipelago.service',
        d09_workspace=dict(path=str(generation.WORKSPACE),initial_digest=generation.object_digest(generation.WORKSPACE)))
    if migration_layout is not None:value.update(version=3,migration_layout=migration_layout)
    dump(policy.CONFIG,value)
    generation.DIRECTORY.mkdir(mode=0o700)
    policy.Policy() # actual strict schema and preimages, no constructor patching
    p=run('systemd-run','--wait','--pipe','--unit=c01-transition','-p','Delegate=yes','-p','DelegateSubgroup=supervisor',
          '/usr/bin/python3','-I','-S','/usr/lib/archipelago/c01_access.py','transition',check=False)
    write(EVIDENCE/'access-transition.log',p.stdout+p.stderr)
    if (policy.STATE/'journal.json').exists():
        write(EVIDENCE/'access-transition-journal.json',(policy.STATE/'journal.json').read_bytes())
    assert p.returncode==0, 'actual access transition failed; retained journal/log'
    operator_route(True)
    if post_retirement is not None:post_retirement()
    run('systemctl','start','archipelago-c01-access.service')
    wait_path('/run/archipelago-c01/runtime-access.sock')
    return {'policy_sha256':sha(policy.CONFIG),'journal_sha256':sha(policy.STATE/'journal.json')}


def runtime_owner():
    run('systemctl','start','archipelago-runtime-broker.service')
    wait_path('/run/archipelago-managed/readiness.sock')
    proof=run('/usr/bin/python3','-B','-c', '''import json,os,stat,sys
sys.path.insert(0,'/usr/lib/archipelago')
import c01_runtime,c01_runtime_view,c01_views
value,digest=c01_runtime.load_config()
lease=c01_runtime_view.RuntimeLease(value['manager_uid'],value['manager_gid'],value['runtime_uid'],value['runtime_gid'],digest)
try:
 mounts=c01_views.PrivateMounts();mounts.cover('/run',0o755)
 lease.install();lease.check()
 meta=os.stat('/run/archipelago-managed/runtime.sock')
 assert stat.S_ISSOCK(meta.st_mode)
 print(json.dumps({'wire_version':2,'device':meta.st_dev,'inode':meta.st_ino,'worker_pid':lease.identity['pid']}))
finally:lease.close()
''')
    receipt=json.loads(proof.stdout)
    dump(EVIDENCE/'runtime-detached-mount.json',receipt)
    return {'broker_main_pid':run('systemctl','show','archipelago-runtime-broker.service','-p','MainPID','--value').stdout.decode().strip(),
            'detached_socket_mount':receipt}


def d09_owner():
    provision=d09_module()
    owner=provision.owner
    launch,_=policy.json_read(policy.LAUNCH)
    runtime,_=policy.json_read(policy.RUNTIME)
    access_journal,_=policy.json_read(policy.STATE/'journal.json')
    assert access_journal['state']=='owner-active'
    incoming=owner.ROOT/'incoming'
    web=owner.DESTINATIONS['frontend']
    # Use actual disposable image content as the original candidate, preserving
    # it in the owner's immutable restore snapshot. Never delete a destination.
    assert web.is_dir()
    for label in ('new','restore'):
        shutil.copytree(web,incoming/(label+'-frontend'))
        shutil.copy2(owner.DESTINATIONS['backend'],incoming/(label+'-backend'))
    with (incoming/'new-backend').open('ab') as stream: stream.write(b'\nC01-disposable-generation-B\n')
    new_launch=dict(launch,native_sha256=sha(incoming/'new-backend'))
    dump(incoming/'new-launch',new_launch)
    dump(incoming/'restore-launch',launch)
    write(incoming/'new-config',Path('/usr/libexec/archipelago-d09-terminal-workspace.conf').read_bytes())
    dump(incoming/'exclusion.json',access_journal)
    effective=owner.effective_configuration()
    effective_sha=hashlib.sha256(effective.encode()).hexdigest()
    projected=json.loads((EVIDENCE/'candidate-systemd-config.json').read_text())
    assert effective_sha==projected['old_sha256']
    binding=dict(schema_version=1,target_binding_digest=hashlib.sha256(generation.canonical(dict(
        boot_id=Path('/proc/sys/kernel/random/boot_id').read_text().strip(),
        access_sha256=sha(policy.CONFIG), runtime_sha256=sha(policy.RUNTIME)))).hexdigest(),
        composition_digest=hashlib.sha256(generation.canonical(dict(launch_sha256=sha(policy.LAUNCH),
            network_sha256=sha(c01_network_owner.CONFIG),native_sha256=sha(owner.DESTINATIONS['backend'])))).hexdigest(),
        admission_epoch_id=secrets.token_hex(16),transaction_id=secrets.token_hex(16),
        grants={role:secrets.token_hex(32) for role in ('UPDATER','AUTOMATIC','MANUAL','GUARD','OBSERVER')},
        artifacts=dict(backend=sha(incoming/'new-backend'),frontend=owner.object_digest(incoming/'new-frontend')),
        restore_artifacts=dict(backend=sha(owner.DESTINATIONS['backend']),frontend=owner.object_digest(web)),
        effective_config={'APPLY':projected['new_sha256'],'RESTORE':effective_sha},
        exclusion_evidence_digest=sha(incoming/'exclusion.json'),helper_digest=sha('/usr/libexec/archipelago-d09-prepare-workspace'),
        prerequisite=dict(grant_digest=secrets.token_hex(32),backend_sha256=sha(owner.DESTINATIONS['backend']),effective_config_digest=effective_sha))
    binding['release_identity']=['fixture-B',[['archipelago','fixture-B',binding['artifacts']['backend'],(incoming/'new-backend').stat().st_size]]]
    binding['c01']=dict(version=1,application=dict(name='c01-manager',uid=61001,gid=61001,home='/var/empty'),
        runtime=dict(name='archipelago',uid=1000,gid=1000,home='/home/archipelago'),runtime_sha256=sha(policy.RUNTIME),
        generations={name:dict(native_sha256=item['native_sha256'],launch_sha256=hashlib.sha256(generation.canonical(item)).hexdigest())
            for name,item in [('old',launch),('new',new_launch),('restore',launch)]},transitions={})
    binding['objects']={name:dict(old=owner.object_digest(path),new=owner.object_digest(incoming/('new-'+name)),
        restore=owner.object_digest(incoming/('restore-'+name))) for name,path in owner.destinations(binding).items()}
    for action,old,new in [('APPLY',launch,new_launch),('RESTORE',new_launch,launch)]:
        record=dict(version=1,policy_sha256=sha(policy.CONFIG),runtime_sha256=sha(policy.RUNTIME),
            transaction_id=binding['transaction_id'],action=action,grant_digest=secrets.token_hex(32),
            from_launch=old,to_launch=new,from_launch_sha256=hashlib.sha256(generation.canonical(old)).hexdigest(),
            to_launch_sha256=hashlib.sha256(generation.canonical(new)).hexdigest(),
            from_workspace_digest=binding['objects']['config']['old' if action=='APPLY' else 'new'],
            to_workspace_digest=binding['objects']['config']['new' if action=='APPLY' else 'restore'])
        history=generation.canonical(record)
        record['external_history_digest']=hashlib.sha256(history).hexdigest()
        write(incoming/('c01-history-'+record['external_history_digest']+'.json'),history)
        raw=generation.canonical(record);digest=hashlib.sha256(raw).hexdigest()
        write(incoming/('c01-generation-'+digest+'.json'),raw)
        binding['c01']['transitions'][action]=digest
    dump(incoming/'admission.json',binding)
    provision.provision(incoming/'admission.json')
    run('systemctl','start','archipelago-d09-lifecycle.service')
    wait_path('/run/archipelago-d09/c01.sock')
    return {'binding_sha256':sha(owner.BINDING),'ledger_sha256':sha(owner.ROOT/'ledger.json')}


def main():
    assert os.geteuid()==0 and Path('/run/.containerenv').exists()
    assert socket.gethostname()=='c01-prerequisite-qualification' and os.environ.get('C01_NORMAL_PID1')=='1'
    results = {}
    operations=[('structures',structures), ('stores',stores), ('epoch_owner',epoch_owner), ('network_owner',network_owner),
                             ('nts_prepare',nts_prepare),
                             ('prepare_owners',prepare_owners), ('role_view_probe',role_view_probe), ('conditional_contract',conditional_contract), ('nts_owner',nts_owner),
                            ('access_owner',access_owner), ('runtime_owner',runtime_owner),
                             ('d09_owner',d09_owner)]
    if os.environ.get('C01_HISTORY_MIGRATION')=='1':
        import runpy
        migration=runpy.run_path(str(PACKAGE/'tests/regression/c01-history-migration.py'))['MigrationFixture'](globals())
        operations=[('structures',structures),('history_seed',migration.seed),('network_owner',network_owner),
                    ('nts_prepare',nts_prepare),('prepare_owners',prepare_owners),
                    ('role_view_probe',role_view_probe),('conditional_contract',conditional_contract),('nts_owner',nts_owner),
                    ('access_owner',lambda:access_owner(migration.import_retired,
                        migration.layout_plan() if os.environ.get('C01_LAYOUT_MIGRATION')=='1' else None)),
                    ('runtime_owner',runtime_owner),('d09_owner',d09_owner)]
    if os.environ.get('C01_RUNNING_SERVICES')=='1':
        import runpy
        running=runpy.run_path(str(PACKAGE/'tests/regression/c01-running-services.py'))
        index=next(i for i,(name,_) in enumerate(operations) if name=='prepare_owners')+1
        operations.insert(index,('running_services',lambda:running['install'](globals())))
        oci=runpy.run_path(str(PACKAGE/'tests/regression/c01-appointed-oci.py'))
        index=next(i for i,(name,_) in enumerate(operations) if name=='runtime_owner')+1
        operations.insert(index,('appointed_oci',lambda:oci['qualify'](globals())))
    if (PACKAGE/'node-cutover-fixture.json').exists():
        import runpy
        custody=runpy.run_path(str(PACKAGE/'tests/regression/node-cutover-custody.py'))
        index=next(i for i,(name,_) in enumerate(operations) if name=='structures')+1
        operations.insert(index,('durable_os_custody',lambda:custody['qualify'](globals())))
    if os.environ.get('C01_POWER_COMPONENT')=='1':
        operations=[(name,operation) for name,operation in operations if name not in ('nts_owner','nts_prepare')]
    for name, operation in operations:
        try:
            result = operation()
            # Structural inputs have no private key bytes; record hashes only.
            results[name] = {'status':'completed', 'result':result if name != 'structures' else
                {'launch_sha256':sha(policy.LAUNCH), 'runtime_sha256':sha(policy.RUNTIME)}}
        except Exception as error:
            write(EVIDENCE / (name + '-error.log'), traceback.format_exc())
            results[name] = {'status':'failed', 'error':str(error)}
            dump(EVIDENCE / 'composition.json', results)
            break  # Never advance an incomplete migration into runtime/activation.
        dump(EVIDENCE / 'composition.json', results)
    print(json.dumps(results), flush=True)
    return 0 if all(item['status']=='completed' for item in results.values()) else 1


if __name__=='__main__':
    raise SystemExit(main())