#!/usr/bin/env python3
"""Native crypto + real bootstrap/D09/OS test. Not candidate qualification.

Reuses the peer lane's guest fixture scaffolding with explicit, checked fixture-
only substitutions. Production helpers/crypto/executor are never monkeypatched.
"""
import hashlib
import json
import os
from pathlib import Path
import subprocess
import socket
import struct
import sys
import time

assert 'QEMU' in Path('/sys/class/dmi/id/sys_vendor').read_text()
assert subprocess.check_output(['findmnt','-n','-o','SOURCE','/'],text=True).strip()=='/dev/vda'
PACKAGE=Path('/opt/ota-source')
sys.path.insert(0,str(PACKAGE/'scripts'))
import os_update_executor as executor
import os_updater_activate as activation
import os_updater_bootstrap as bootstrap
assert executor.ROOT==activation.OS_ROOT==Path('/var/lib/archipelago-os-update'), 'correct root custody prerequisite'
assert 'var/lib/archipelago-os-update/transactions' in bootstrap.OS_TRANSACTION_ROOTS
assert not (PACKAGE/'accepted-old-archipelago').exists(), 'isolated fixture must not claim accepted native candidate'
fixture=Path('/opt/os-ota-crypto-native')
proof=json.loads(Path('/opt/os-ota-crypto-fixture.json').read_text())
assert proof['scope']=='ISOLATED_COMPILETIME_TEST_ROOT_NOT_CANDIDATE'
assert executor.file_hash(fixture)==proof['binary_sha256']
fixture.chmod(0o755)
keys=Path('/root/os-ota-crypto-fixture')
keys.mkdir(mode=0o700)
Path('/root/os-ota-crypto.seed').rename(keys/'test.seed')
(keys/'test.seed').chmod(0o600)
(keys/'wrong.seed').write_bytes(os.urandom(32))
(keys/'wrong.seed').chmod(0o600)

def preservation_probes(context):
    account=context['account']
    home=Path('/home/archipelago')
    probes=[home/'Work/operator-fixture.txt',home/'.local/state/archipelago/sessions/history-fixture.txt',
            Path('/etc/systemd/system/archipelago.service.d/90-operator-fixture.conf')]
    for path in probes[:2]:
        path.parent.mkdir(parents=True,exist_ok=True)
        path.write_text('preserve exact fixture bytes, identity and intent\n')
        path.chmod(0o600)
        for parent in (path,*path.parents):
            if parent==home.parent:break
            os.chown(parent,account.pw_uid,account.pw_gid)
    probes[2].parent.mkdir(parents=True,exist_ok=True)
    probes[2].write_text('[Service]\nEnvironment=OS_OTA_OPERATOR_CANARY=retained\n')
    context['crypto_preservation_paths']=probes
    context['crypto_preservation_before']=[path_identity(p) for p in probes]

def path_identity(path):
    st=path.stat()
    return [str(path),st.st_ino,st.st_mode,st.st_uid,st.st_gid,executor.file_hash(path)]

def capture_owner(context):
    pid=int(subprocess.check_output(['systemctl','show','archipelago-d09-lifecycle.service','-p','MainPID','--value'],text=True))
    context['crypto_owner_before']=context['owner'].process(pid)
    context['crypto_epoch_before']=json.loads((context['owner'].ROOT/'ledger.json').read_text())['epoch']
    context['crypto_root_uuid']=subprocess.check_output(['blkid','-s','UUID','-o','value','/dev/vda'],text=True).strip()

def crypto_appointment(context):
    binding=context['binding']
    source=context['context']['source']
    manifest=source/'manifest.json'
    unsigned=source/'unsigned-fixture-manifest.json'
    manifest.rename(unsigned)
    subprocess.run([str(fixture),'fixture-sign',str(keys/'test.seed'),str(unsigned),str(manifest)],check=True)
    checks=[]
    def verify(name,path,accept,env=None):
        r=subprocess.run([str(fixture),'ceremony','verify',str(path)],capture_output=True,text=True,env=env)
        assert (r.returncode==0)==accept,(name,r.stdout,r.stderr)
        checks.append(dict(name=name,exit=r.returncode,stdout=r.stdout,stderr=r.stderr))
    verify('genuine-fixture-signature',manifest,True)
    verify('unsigned-refused',unsigned,False)
    tamper=json.loads(manifest.read_text());tamper['qualification']='tampered after signature'
    (source/'tampered.json').write_text(json.dumps(tamper))
    verify('tampered-signature-refused',source/'tampered.json',False)
    subprocess.run([str(fixture),'fixture-sign',str(keys/'wrong.seed'),str(unsigned),str(source/'wrong-key.json')],check=True)
    verify('valid-signature-wrong-root-refused',source/'wrong-key.json',False)
    env=dict(os.environ,ARCHY_RELEASE_ROOT_PUBKEY='00'*32)
    verify('runtime-env-cannot-replace-compiled-test-pin',manifest,True,env)
    production=Path('/opt/os-ota-native')
    if production.exists():
        production.chmod(0o755)
        r=subprocess.run([str(production),'ceremony','verify',str(manifest)],capture_output=True,text=True)
        assert r.returncode!=0 and 'pinned release-root anchor' in r.stderr,r.stderr
        checks.append(dict(name='production-root-refuses-test-signature',exit=r.returncode,stderr=r.stderr))
    binding['os_update']['manifest_sha256']=executor.file_hash(manifest)
    # Planner adapter is retained as evidence, replaced only on this fresh VM by
    # genuine native verification using the isolated test anchor.
    if executor.VERIFIER.exists():
        assert 'ROOT_APPOINTED_EXACT_FIXTURE_ONLY_NOT_PRODUCTION_SIGNATURE' in executor.VERIFIER.read_text()
        executor.VERIFIER.rename(source/'retained-planner-exact-pin-adapter.py')
    import shutil
    shutil.copyfile(fixture,executor.VERIFIER)
    executor.VERIFIER.chmod(0o755)
    binding['os_update']['verifier_sha256']=executor.file_hash(executor.VERIFIER)
    assert binding['os_update']['verifier_sha256']==binding['prerequisite']['backend_sha256']
    assert not (executor.ROOT/'verifier-custody.json').exists(), 'custody must be established by actual provisioner'
    (Path('/root/fixture-verifier-appointment.json')).write_text(json.dumps(dict(
        fixture_only=True,cryptographic_verification=True,production_signature_approval=False,proof=proof)))
    (Path('/var/log/os-ota-stage')/'native-crypto-checks.json').write_text(json.dumps(checks,indent=2)+'\n')
    print('GENUINE_NATIVE_CRYPTO_CHECKS',json.dumps(checks),flush=True)

source=PACKAGE/'scripts/tests/os_updater_activation_guest.py'
raw=source.read_text()
original_hash=hashlib.sha256(raw.encode()).hexdigest()
old='python = Path(sys.executable).resolve().read_bytes()'
assert raw.count(old)==1
raw=raw.replace(old,"python = Path('/opt/os-ota-crypto-native').read_bytes()")
old='ExecStart=/usr/local/bin/archipelago /opt/ota-source/requester.py'
assert old in raw
# NATIVE=False; replacing the base unit command does not activate the separate
# accepted-candidate branch of the peer fixture.
raw=raw.replace(old,'ExecStart=/usr/local/bin/archipelago')
old='NoNewPrivileges=yes\nProtectSystem=strict'
assert raw.count(old)==1
raw=raw.replace(old,'RuntimeDirectory=d09-fixture-client\nRuntimeDirectoryMode=0700\nRuntimeDirectoryPreserve=yes\n'+old)
start=raw.index('# A tiny native exact-pin adapter in the disposable guest')
end=raw.index("write(incoming / 'admission.json', wire.canonical(binding))",start)
raw=raw[:start]+'crypto_appointment(globals())\n'+raw[end:]
raw=raw.replace('ROOT-APPOINTED-EXACT-FIXTURE-NOT-PRODUCTION-SIGNATURE','NATIVE_CRYPTO_COMPILETIME_TEST_ROOT_NOT_CANDIDATE')
marker='# A separate pre-existing fixture admin represents the independently verified'
assert raw.count(marker)==1
raw=raw.replace(marker,'preservation_probes(globals())\n'+marker)
marker="assert result['state'] == 'D09_PREREQUISITE_ACTIVATED'"
assert raw.count(marker)==1
raw=raw.replace(marker,marker+'\ncapture_owner(globals())')
context=dict(__name__='isolated_crypto_bootstrap',__file__=str(source),crypto_appointment=crypto_appointment,
             preservation_probes=preservation_probes,capture_owner=capture_owner)
transformed=Path('/root/os-ota-crypto-transformed-guest.py')
transformed.write_text(raw)
Path('/var/log/journal').mkdir(exist_ok=True)
subprocess.run(['journalctl','--flush'],check=True)
try:
    exec(compile(raw,str(transformed),'exec'),context)
finally:
    evidence=Path('/var/log/os-ota-stage')
    evidence.mkdir(exist_ok=True)
    units=['archipelago-d09-lifecycle.service','archipelago.service']
    ledger=Path('/var/lib/archipelago-d09/ledger.json')
    if ledger.exists():
        value=json.loads(ledger.read_text())
        (evidence/'retained-d09-ledger.json').write_bytes(ledger.read_bytes())
        if value.get('execution'):
            units.append(value['execution']['unit'])
    command=['journalctl','--no-pager','-o','short-monotonic']
    for unit in units:command+=['-u',unit]
    r=subprocess.run(command,capture_output=True,text=True)
    (evidence/'crypto-d09-journal.log').write_text(r.stdout+r.stderr)
    r=subprocess.run(['systemctl','show',*units],capture_output=True,text=True)
    (evidence/'crypto-d09-units.log').write_text(r.stdout+r.stderr)
owner=context['owner']
deadline=time.monotonic()+120
while time.monotonic()<deadline:
    value=json.loads((owner.ROOT/'ledger.json').read_text())
    phase=(value.get('receipt') or {}).get('phase')
    if phase in ('AWAITING_OBSERVATION','COMPLETE'):
        break
    if phase=='CONFLICT':
        raise AssertionError(value['receipt'])
    time.sleep(.2)
else:
    raise TimeoutError('D09 worker did not finish app/config activation after OS apply')
# Actual namespace/PTY and exact preexisting file observations, then actual
# root-observer ACK. COMPLETE remains distinct from the pending OS reboot.
import d09_wire as wire
import d09_store as store
receipt=value['receipt']
pid=owner.service()['MainPID']
subprocess.run(['nsenter','-t',pid,'-m','--','runuser','-u','archipelago','--','bash','-c',
                'cd /home/archipelago/Work && script -q -c pwd /dev/null'],check=True)
for directory in ('/home/archipelago/Work','/home/archipelago/.local/state/archipelago/sessions'):
    subprocess.run(['nsenter','-t',pid,'-m','--','runuser','-u','archipelago','--','touch',directory+'/os-ota-observed-write'],check=True)
for path in ('/usr/local/bin/os-ota-observed-denied','/opt/archipelago/web-ui/os-ota-observed-denied'):
    assert subprocess.run(['nsenter','-t',pid,'-m','--','runuser','-u','archipelago','--','touch',path],capture_output=True).returncode!=0
after=[path_identity(p) for p in context['crypto_preservation_paths']]
assert after==context['crypto_preservation_before']
base=dict(transaction_id=receipt['transaction_id'],observed_invocation_id=receipt['observed_invocation_id'])
namespace=dict(base,qualification='isolated native requester and real OS PTY; not product terminal/C01 UAT',
               default_terminal_created=True,default_terminal_workspace='/home/archipelago/Work',
               namespace_writable=['/home/archipelago/Work','/home/archipelago/.local/state/archipelago/sessions'])
preservation=dict(base,preserved={name:dict(before=wire.digest(wire.canonical(before)),after=wire.digest(wire.canonical(now)))
    for name,before,now in zip(('work','history','custom_fragments'),context['crypto_preservation_before'],after)})
def publish(value):
    raw=wire.canonical(value);hashed=wire.digest(raw)
    store.atomic(owner.ROOT/'observations'/(hashed+'.json'),raw)
    return hashed
observation={key:receipt[key] for key in ('transaction_id','target_binding_digest','composition_digest',
    'observed_invocation_id','effective_config_observation_digest','config_current_digest')}
observation.update(namespace_observation_digest=publish(namespace),preservation_observation_digest=publish(preservation))
binding=context['binding']
request=dict(schema_version=1,operation='ACK_OBSERVATION',route='OBSERVER',reason='OBSERVATION_READY',
    transaction_id=receipt['transaction_id'],request_id=os.urandom(16).hex(),target_binding_digest=receipt['target_binding_digest'],
    composition_digest=receipt['composition_digest'],grant_digest=binding['grants']['OBSERVER'],
    expected_sequence=value['sequence'],requester_invocation_id=None,observer_receipt_digest=publish(observation),
    selected_release_digest=None,staging_generation=None)
def exchange():
    with socket.socket(socket.AF_UNIX,socket.SOCK_STREAM) as conn:
        conn.settimeout(150);conn.connect(str(owner.SOCKET))
        assert struct.unpack('3i',conn.getsockopt(socket.SOL_SOCKET,socket.SO_PEERCRED,12))[1]==0
        conn.sendall(wire.canonical(request));conn.shutdown(socket.SHUT_WR)
        raw=b''
        while part:=conn.recv(8192):raw+=part
        return wire.parse(raw)
assert exchange()['status']=='ACCEPTED'
deadline=time.monotonic()+90
while time.monotonic()<deadline:
    value=json.loads((owner.ROOT/'ledger.json').read_text())
    if value['receipt']['phase']=='COMPLETE':break
    assert value['receipt']['phase']!='CONFLICT',value['receipt']
    time.sleep(.2)
else:raise TimeoutError('same-boot userspace COMPLETE')
assert value['receipt']['outcome']=='COMPLETE'
deadline=time.monotonic()+30
while time.monotonic()<deadline:
    active=subprocess.check_output(['systemctl','show',value['execution']['unit'],'-p','ActiveState','--value'],text=True).strip()
    if active in ('inactive','failed'):break
    time.sleep(.1)
else:raise TimeoutError('completed observation worker drain')
head=(owner.ROOT/'ledger.json').read_bytes()
assert exchange()['status']=='ACCEPTED'
assert (owner.ROOT/'ledger.json').read_bytes()==head
phase='COMPLETE'
owner_after=owner.process(context['crypto_owner_before']['pid'])
assert owner_after==context['crypto_owner_before']
assert value['epoch']==context['crypto_epoch_before']
assert os.stat('/var/lib/archipelago').st_uid==1000
assert executor.ROOT.stat().st_uid==0 and executor.ROOT.stat().st_mode & 0o022==0
subprocess.run(['runuser','-u','archipelago','--','touch','/var/lib/archipelago/os-ota-custody-positive-probe'],check=True)
denials=[]
for args in (['touch',str(executor.ROOT/'app-write-denied')],
             ['python3','-c',"import os;os.rename('/var/lib/archipelago-os-update','/var/lib/app-rename-denied')"],
             ['python3','-c',"open('/root/os-ota-crypto-fixture/test.seed','rb').read()"]):
    r=subprocess.run(['runuser','-u','archipelago','--',*args],capture_output=True,text=True)
    assert r.returncode!=0,args
    denials.append(dict(command=args,exit=r.returncode))
evidence=Path('/var/log/os-ota-stage')
before=executor.inventory()
binding=context['binding']
status=executor.Executor(binding['os_update']).status
assert status['phase']=='REBOOT_REQUIRED'
assert executor.stage(binding,'APPLY','apply')==status
assert executor.inventory()==before
wallet=Path('/var/lib/archipelago/ota-synthetic-canaries/wallet.db')
with wallet.open('a') as f:f.write('FORWARD-AFTER-CRYPTO-D09-sequence-18\n')
forward={path:executor.file_hash(Path(path)) for path in context['preserved']}
result=dict(passed=True,production_approved=False,candidate_artifact_qualified=False,
    genuine_native_crypto=True,fixture_proof=proof,bootstrap_fixture_source_sha256=original_hash,
    transformed_fixture_sha256=executor.file_hash(transformed),
    production_helpers_unmodified=True,os_root=str(executor.ROOT),app_data_uid=1000,
    denied_app_mutations=denials,d09_phase=phase,os_phase=status['phase'],
    same_boot_userspace_complete=True,observer_ack_idempotent=True,owner_lifetime_before=context['crypto_owner_before'],
    owner_lifetime_after=owner_after,manager_events=value.get('manager_events',[]),
    signed_manager_reexec=context['context']['plan']['manager_reexec'],
    workspace_preservation=context['crypto_preservation_before'],
    root_filesystem_uuid=context['crypto_root_uuid'],in_place_package_route=True,
    target_kernel=context['context']['plan']['boot']['expected_release'],forward_preserved=forward)
import importlib.util
provision=context['provision']
installed={name:executor.file_hash(Path('/usr/libexec')/name) for name in provision.MODULES}
for name,hashed in installed.items():assert hashed==executor.file_hash(PACKAGE/'scripts'/name),name
result['installed_module_sha256']=installed
result['old_boot_retained']=all(executor.file_hash(path)==sha for path,sha in context['context']['plan']['boot']['retained_files'].items() if path!='/boot/grub/grub.cfg')
assert result['old_boot_retained']
(evidence/'same-boot-complete-ledger.json').write_bytes(head)
(evidence/'same-boot-complete-observation.json').write_bytes(wire.canonical(observation))
(evidence/'crypto-result.json').write_text(json.dumps(result,indent=2)+'\n')
print('CRYPTO_BOOTSTRAP_OS_PASS',json.dumps(result),flush=True)