#!/usr/bin/python3
"""Real native FD10 component lane under PID1, never installed C01 acceptance.

Executes the compiled production Rust transport/extraction with a native root
fixture supervisor and native non-root child. Full C01 clock/storage/admission
remains separately refused. Generation inspection uses the production controller
with synthetic external appointments, not a fake healthy-process response.
"""
import hashlib
import importlib.util
import json
import os
from pathlib import Path
import pwd
import secrets
import shutil
import socket
import subprocess
import sys
import time

assert os.geteuid()==0 and socket.gethostname()=='d09-native-qualification'
assert Path('/run/.containerenv').exists() and os.environ.get('D09_NATIVE_PID1')=='1'
PACKAGE=Path('/opt/d09-native')
for directory,_,files in os.walk(PACKAGE):
    for path in [Path(directory),*(Path(directory)/name for name in files)]:path.chmod(path.stat().st_mode & 0o755)

def run(*args,check=True):
    r=subprocess.run(args,capture_output=True,text=True,timeout=60)
    if check and r.returncode:raise AssertionError((args,r.returncode,r.stdout,r.stderr))
    return r
def write(path,value,mode=0o644):
    path=Path(path);path.parent.mkdir(parents=True,exist_ok=True);path.write_bytes(value.encode() if isinstance(value,str) else value);path.chmod(mode)
def wait(predicate,label,seconds=100):
    end=time.monotonic()+seconds
    while time.monotonic()<end:
        value=predicate()
        if value:return value
        time.sleep(.1)
    raise AssertionError('timed out: '+label)
def sha(path):
    with Path(path).open('rb') as source:return hashlib.file_digest(source,'sha256').hexdigest()

run('systemctl','stop','archipelago.service',check=False)
run('tar','-xf',str(PACKAGE/'native-test.tar'),'-C','/')
run('tar','-xf',str(PACKAGE/'owners.tar'),'-C','/')
for path,expected in json.loads((PACKAGE/'native-artifact-sha256.json').read_text()).items():
    assert sha('/'+path)==expected['sha256'],path
for path,expected in json.loads(Path('/D09-C01-SOURCE-SHA256.json').read_text()).items():
    assert sha('/'+path)==expected,path
sys.path.insert(0,str(PACKAGE/'scripts'))
import d09_wire as wire
import d09_store as store
spec=importlib.util.spec_from_file_location('provision',PACKAGE/'scripts/provision-d09-lifecycle.py')
provision=importlib.util.module_from_spec(spec);spec.loader.exec_module(provision)
owner=provision.owner
import c01_access_generation as generation
import c01_access_policy as policy
root=Path('/var/lib/d09-native');root.mkdir(mode=0o755)
write('/etc/nsswitch.conf','passwd: files\ngroup: files\nshadow: files\nhosts: files dns\n')
run('groupadd','--gid','61001','c01-native-app')
run('useradd','--uid','61001','--gid','61001','--no-create-home','--home-dir','/var/empty','--shell','/usr/sbin/nologin','c01-native-app')
run('usermod','-G','','archipelago')
run('usermod','-s','/usr/sbin/nologin','-L','archipelago')
(root/'manager').mkdir(mode=0o700);os.chown(root/'manager',61001,61001)
run('ip','netns','add','archipelago-managed')
run('nsenter','--net=/run/netns/archipelago-managed','--','ip','link','set','lo','up')
for p in (owner.BINDING.parent,owner.ROOT):
    if p.exists():raise AssertionError('fresh authority required')
web=owner.DESTINATIONS['frontend']
if web.exists():shutil.rmtree(web)
os.chown(web.parent,0,0);web.parent.chmod(0o755)
write(web/'index.html','native handoff fixture\n')
write(web/'archipelago-runtime/image-recipe/configs/nginx-archipelago.conf',(PACKAGE/'image-recipe/configs/nginx-archipelago.conf').read_bytes())
write('/etc/systemd/system/archipelago.service','''[Unit]
Description=Native protected D09 handoff component fixture
[Service]
Type=notify
NotifyAccess=main
User=root
Group=root
NoNewPrivileges=yes
Delegate=yes
DelegateSubgroup=supervisor
NetworkNamespacePath=/run/netns/archipelago-managed
ExecStart=/usr/local/bin/archipelago --ignored --exact d09_transport::tests::pid1_supervisor --nocapture --test-threads=1
Environment=D09_NATIVE_PID1=1
TimeoutStartSec=150
Restart=no
''')
if owner.DESTINATIONS['config'].parent.exists():shutil.rmtree(owner.DESTINATIONS['config'].parent)
run('systemctl','daemon-reload')
provision.install_package()
old_native=sha(owner.DESTINATIONS['backend'])
runtime={'version':2,'manager_uid':61001,'manager_gid':61001,'runtime_uid':1000,'runtime_gid':1000,
    'authority_uid':61002,'authority_gid':61002,'operator_uid':1001,'operator_gid':1001,
    'runtime_user':'archipelago','home':'/home/archipelago','home_backing':str(root/'runtime-home'),
    'runtime_dir':'/run/user/1000','runtime_data_dir':str(root/'runtime-data'),'graphroot':str(root/'graphroot'),
    'subuids':[[int(row.split(':')[1]),int(row.split(':')[2])] for row in Path('/etc/subuid').read_text().splitlines() if row.startswith('archipelago:')],
    'subgids':[[int(row.split(':')[1]),int(row.split(':')[2])] for row in Path('/etc/subgid').read_text().splitlines() if row.startswith('archipelago:')],
    'user_cgroup':'/user.slice/user-1000.slice/user@1000.service',
    'private_parents':[str(root/('private-'+role)) for role in ('manager','snapshot','authority')],
    'workspace_helper_sha256':sha('/usr/libexec/archipelago-d09-prepare-workspace')}
for name in ('runtime-home','runtime-data','graphroot'):
    (root/name).mkdir(mode=0o700);os.chown(root/name,1000,1000)
for path in runtime['private_parents']:Path(path).mkdir(mode=0o700)
write('/etc/archipelago-c01/runtime.json',wire.canonical(runtime),0o600)
launch={'version':3,'native_sha256':old_native,'application_user':'c01-native-app','authority_user':'c01-native-authority','runtime_uid':1000,'runtime_gid':1000,
    **{k:str(root/k) for k in ('manager_backing','snapshot_backing','authority_backing','snapshot_key','authority_key','platform_profile','proxy_path')},
    'runtime_data_dir':runtime['runtime_data_dir'],'binding':{'local_identity':[1]*32,'incarnation':[2]*32,'reader_schema':2,'writer_schema':2},
    'platform_profile_sha256':'f'*64,'clock_domain':'f'*32,'host_ip':'192.0.2.1'}
write(generation.policy.LAUNCH,wire.canonical(launch),0o600)
base={'version':2,'launch_sha256':sha(generation.policy.LAUNCH),'runtime_sha256':sha('/etc/archipelago-c01/runtime.json'),
    'd09_workspace':{'path':str(generation.WORKSPACE),'initial_digest':wire.digest(b'ABSENT')}}
write(policy.CONFIG,wire.canonical(base),0o600)
policy.STATE.mkdir(parents=True,exist_ok=True,mode=0o700)
generation.DIRECTORY.mkdir(mode=0o700)
Path('/run/archipelago-c01').mkdir(mode=0o755,exist_ok=True)
# This lane exercises only the production generation controller, not the full
# access-owner migration/lease grant. Its input is explicitly synthetic.
write(PACKAGE/'generation-controller.py','''import json,os,pathlib,socket,struct,sys
sys.path.insert(0,'/opt/d09-native/scripts')
import c01_access_generation as g,c01_access_policy as p,c01_access_kernel as k
v,h=p.json_read(p.CONFIG);launch,d=p.json_read(p.LAUNCH);controller=g.Generations(v,h,launch,d)
s=socket.socket(socket.AF_UNIX,socket.SOCK_SEQPACKET);s.bind('/run/archipelago-c01/runtime-access.sock');os.chmod('/run/archipelago-c01/runtime-access.sock',0o600);s.listen(4)
while True:
 c,_=s.accept();c.settimeout(2)
 try:
  pid,uid,gid=struct.unpack('3i',c.getsockopt(socket.SOL_SOCKET,socket.SO_PEERCRED,12))
  assert (uid,gid)==(0,0) and k.process(pid)['group']=='/sys/fs/cgroup/system.slice/archipelago-d09-lifecycle.service'
  r=json.loads(c.recv(8192));assert r['method']=='inspect'
  c.sendall(g.canonical(controller.handle(r,False)))
 finally:c.close()
''')
run('systemd-run','--unit=c01-generation-fixture','--property=Type=exec','/usr/bin/python3','-B',str(PACKAGE/'generation-controller.py'))
wait(lambda:Path('/run/archipelago-c01/runtime-access.sock').exists(),'generation controller')
incoming=owner.ROOT/'incoming'
for generation_name in ('new','restore'):
    shutil.copytree(web,incoming/(generation_name+'-frontend'))
    shutil.copy2(owner.DESTINATIONS['backend'],incoming/(generation_name+'-backend'))
    if generation_name=='new':
        with (incoming/'new-backend').open('ab') as stream:stream.write(b'\nB\n')
new_launch=dict(launch,native_sha256=sha(incoming/'new-backend'))
write(incoming/'new-launch',wire.canonical(new_launch),0o600)
write(incoming/'restore-launch',wire.canonical(launch),0o600)
write(incoming/'new-config',Path('/usr/libexec/archipelago-d09-terminal-workspace.conf').read_bytes())
write(incoming/'exclusion.json',wire.canonical({'scope':'synthetic native descriptor component fixture; NOT migration exclusion'}),0o600)
old_effective=wire.digest(owner.effective_configuration().encode())
transaction=secrets.token_hex(16)
binding=dict(schema_version=1,target_binding_digest=secrets.token_hex(32),composition_digest=secrets.token_hex(32),
    admission_epoch_id=secrets.token_hex(16),transaction_id=transaction,
    grants={r:secrets.token_hex(32) for r in ('UPDATER','AUTOMATIC','MANUAL','GUARD','OBSERVER')},
    artifacts={'backend':sha(incoming/'new-backend'),'frontend':owner.object_digest(incoming/'new-frontend')},
    restore_artifacts={'backend':old_native,'frontend':owner.object_digest(web)},effective_config={'APPLY':old_effective,'RESTORE':old_effective},
    exclusion_evidence_digest=sha(incoming/'exclusion.json'),helper_digest=sha('/usr/libexec/archipelago-d09-prepare-workspace'),
    prerequisite={'grant_digest':secrets.token_hex(32),'backend_sha256':old_native,'effective_config_digest':old_effective})
binding['release_identity']=['B',[['archipelago','B',binding['artifacts']['backend'],(incoming/'new-backend').stat().st_size]]]
binding['c01']={'version':1,'application':{'name':'c01-native-app','uid':61001,'gid':61001,'home':'/var/empty'},
    'runtime':{'name':'archipelago','uid':1000,'gid':1000,'home':'/home/archipelago'},'runtime_sha256':base['runtime_sha256'],
    'generations':{name:{'native_sha256':data['native_sha256'],'launch_sha256':wire.digest(wire.canonical(data))}
        for name,data in [('old',launch),('new',new_launch),('restore',launch)]},'transitions':{}}
binding['objects']={name:{'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=base['runtime_sha256'],transaction_id=transaction,action=action,
        grant_digest=secrets.token_hex(32),from_launch=old,to_launch=new,from_launch_sha256=wire.digest(wire.canonical(old)),
        to_launch_sha256=wire.digest(wire.canonical(new)),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=wire.canonical(record);record['external_history_digest']=wire.digest(history)
    write(incoming/('c01-history-'+record['external_history_digest']+'.json'),history,0o600)
    raw=wire.canonical(record);hashed=wire.digest(raw);binding['c01']['transitions'][action]=hashed
    write(incoming/('c01-generation-'+hashed+'.json'),raw,0o600)
write(incoming/'admission.json',wire.canonical(binding),0o600)
provision.provision(incoming/'admission.json')
run('systemctl','start','archipelago-d09-lifecycle.service')
wait(lambda:owner.PRIVATE.exists() and Path('/run/archipelago-d09/c01.sock').exists(),'independent listeners')
started=time.monotonic()
result=provision.activate_prerequisite()
assert time.monotonic()-started<10 and not result['prerequisite_started']
wait(lambda:(root/'native-proof.json').exists(),'actual native FD10 HELLO/inspect')
proof=json.loads((root/'native-proof.json').read_text())
assert proof['native_hello'] and proof['python_exec_denied'] and proof['host_run_hidden']
assert owner.PERMIT.exists() and owner.service()['ActiveState']=='activating'
before=json.loads(store.read(owner.ROOT/'ledger.json'))
assert before['prerequisite']['state']=='START_INTENT' and before['receipt'] is None
wait(lambda:json.loads(store.read(owner.ROOT/'ledger.json'))['prerequisite']['state']=='STARTED','delayed native READY',70)
assert time.monotonic()-started>=40 and not owner.PERMIT.exists()
assert json.loads(store.read(owner.ROOT/'ledger.json'))['grants']==[binding['prerequisite']['grant_digest']]
print('PASS actual native root MainPID -> native nonroot app FD10: kernel credentials, HELLO, no Python exec, private /run, wrong selection refusal',flush=True)
print('PASS real PID1 startup >40 seconds: one grant/start intent, bounded pre-READY inspect, permit retained until native READY',flush=True)
first_invocation=owner.service()['InvocationID']
original_ledger=store.read(owner.ROOT/'ledger.json')
run('systemctl','stop','archipelago-d09-lifecycle.service')
wait(lambda:owner.service()['MainPID']=='0','owner-loss managed retirement')
print('PASS owner loss retires real native service; no disk-seeded transport authority',flush=True)
(root/'native-proof.json').rename(root/'native-proof-first.json')
run('systemctl','start','archipelago-d09-lifecycle.service')
def unknown():
    response=run('/usr/libexec/archipelago-d09ctl','inspect',check=False)
    return response.returncode==0 and json.loads(response.stdout)['state']=='UNKNOWN'
wait(unknown,'fresh owner refuses disk continuity')
assert store.read(owner.ROOT/'ledger.json')==original_ledger
v=json.loads(original_ledger)
index={k:binding[k] for k in ('target_binding_digest','composition_digest','transaction_id')}
probe=owner.Owner.__new__(owner.Owner)
record=dict(index,schema_version=1,allowed_action='REACTIVATE_PREREQUISITE',
    predecessor_epoch_id=v['epoch'],head_sequence=v['sequence'],observed_boot_id=owner.boot_id(),
    preserved_head_digest=wire.digest(original_ledger),
    preserved_journal_digest=wire.digest(wire.canonical({'receipts':v['journal'],'intents':v['intents']})),
    preserved_consumed_history_digest=wire.digest(wire.canonical(owner.consumption_history(v))),
    preserved_target_index_digest=wire.digest(wire.canonical(index)),disposition_digest=probe.drained_disposition(v),
    new_epoch_id=secrets.token_hex(16),new_grant_digest=secrets.token_hex(32),execution_grant_digest=secrets.token_hex(32),
    external_history_sequence=1)
history=wire.canonical(dict(record,sequence=1));record['external_history_digest']=wire.digest(history)
external=PACKAGE/'external';external.mkdir(mode=0o700)
write(external/'history.json',history,0o600);write(external/'reactivation.json',wire.canonical(record),0o600)
started=time.monotonic()
provision.readmit(external/'reactivation.json',external/'history.json',activation=True)
assert time.monotonic()-started<10
wait(lambda:(root/'native-proof.json').exists(),'new native FD10 after readmission')
assert owner.PERMIT.exists() and owner.service()['ActiveState']=='activating'
invocation=owner.service()['InvocationID'];assert invocation!=first_invocation
before_retry=store.read(owner.ROOT/'ledger.json')
provision.readmit(external/'reactivation.json',external/'history.json',activation=True)
assert store.read(owner.ROOT/'ledger.json')==before_retry and owner.service()['InvocationID']==invocation
wait(lambda:json.loads(store.read(owner.ROOT/'ledger.json'))['prerequisite']['state']=='STARTED','readmitted native READY',70)
assert time.monotonic()-started>=40 and not owner.PERMIT.exists()
assert len(json.loads(store.read(owner.ROOT/'ledger.json'))['grants'])==3
print('PASS external prerequisite readmission >40 seconds: new native FD10, UNKNOWN before authority, identical-record retry nonactivating, permit held through READY',flush=True)