"""Installed-VM D09 integration using the shipped client, owner and worker.

The requester daemon is an inert fixture. This proves the real privileged D09
route, not native application/C01 admission or production release signatures.
No production function, command, APT operation or owner response is replaced.
"""
import fcntl
import importlib.util
import json
import os
from pathlib import Path
import pwd
import secrets
import shutil
import subprocess
import sys
import time

PACKAGE=Path('/opt/ota-source')
EVIDENCE=Path('/var/log/os-ota-stage')
CONTROL=Path('/run/os-ota-d09-client')


def run(*args,check=True):
    result=subprocess.run(args,capture_output=True,text=True,timeout=90)
    print('D09_COMMAND',args,result.returncode,result.stdout,result.stderr,flush=True)
    if check:
        result.check_returncode()
    return result


def write(path,raw,mode=0o644):
    path=Path(path)
    path.parent.mkdir(parents=True,exist_ok=True)
    path.write_bytes(raw.encode() if isinstance(raw,str) else raw)
    path.chmod(mode)


def wait(predicate,label,seconds=1200):
    end=time.monotonic()+seconds
    while time.monotonic()<end:
        value=predicate()
        if value:
            return value
        time.sleep(.2)
    raise TimeoutError(label)


def setup():
    assert 'QEMU' in Path('/sys/class/dmi/id/sys_vendor').read_text()
    assert run('findmnt','-n','-o','SOURCE','/').stdout.strip()=='/dev/vda'
    sys.path.insert(0,str(PACKAGE/'scripts'))
    import d09_wire as wire
    import d09_store as store
    spec=importlib.util.spec_from_file_location('os_vm_provision',PACKAGE/'scripts/provision-d09-lifecycle.py')
    provision=importlib.util.module_from_spec(spec)
    spec.loader.exec_module(provision)
    owner=provision.owner
    assert not owner.BINDING.exists() and not (owner.ROOT/'ledger.json').exists(), 'fresh D09 authority required; use independent disk'
    assert all(name in provision.MODULES for name in ('os_update_executor.py','os_update_plan.py','os_update_repository.py')), 'actual provisioner must package OS modules'
    try:
        account=pwd.getpwnam('archipelago')
    except KeyError:
        run('useradd','--uid','1000','--create-home','--home-dir','/home/archipelago','--shell','/usr/sbin/nologin','archipelago')
        account=pwd.getpwnam('archipelago')
    run('usermod','-G','','archipelago')
    run('systemctl','stop','archipelago.service',check=False)
    CONTROL.mkdir(mode=0o700)
    os.chown(CONTROL,account.pw_uid,account.pw_gid)
    cache=Path('/var/lib/archipelago/d09-requests')
    cache.mkdir(mode=0o700)
    os.chown(cache,account.pw_uid,account.pw_gid)
    home=Path('/home/archipelago')
    home.mkdir(exist_ok=True)
    os.chown(home,account.pw_uid,account.pw_gid)
    requester='''import json,pathlib,subprocess,time
p=pathlib.Path('/run/os-ota-d09-client')
while True:
 for request in p.glob('request-*.json'):
  output=request.with_name(request.name.replace('request-','result-'))
  if output.exists():continue
  value=json.loads(request.read_text())
  r=subprocess.run(['/usr/libexec/archipelago-d09ctl',*value['args']],input=value.get('stdin',''),capture_output=True,text=True,timeout=25)
  output.write_text(json.dumps({'exit':r.returncode,'stdout':r.stdout,'stderr':r.stderr}))
 time.sleep(.1)
'''
    write(PACKAGE/'d09-os-requester.py',requester)
    python=Path(sys.executable).resolve().read_bytes()
    write(owner.DESTINATIONS['backend'],python,0o755)
    web=owner.DESTINATIONS['frontend']
    assert not web.exists(), 'preserve unexpected frontend'
    write(web/'index.html','root-appointed OS OTA requester fixture; not production frontend\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=Private installed OS OTA updater requester fixture
[Service]
Type=exec
User=archipelago
ExecStart=/usr/local/bin/archipelago /opt/ota-source/d09-os-requester.py
NoNewPrivileges=yes
ProtectSystem=strict
ProtectHome=read-only
ReadWritePaths=/run/os-ota-d09-client /var/lib/archipelago/d09-requests
Restart=no
''')
    run('systemctl','daemon-reload')
    provision.install_package()
    denials=[]
    for command in (['touch','/usr/local/bin/os-ota-denied'],['touch',str(web/'denied')],['sudo','-n','true']):
        result=run('runuser','-u','archipelago','--',*command,check=False)
        assert result.returncode!=0
        denials.append(dict(command=command,exit=result.returncode))
    incoming=owner.ROOT/'incoming'
    exclusion=wire.canonical(dict(scope='private VM fixture only',observed_denials=denials))
    store.atomic(incoming/'exclusion.json',exclusion)
    for generation in ('new','restore'):
        shutil.copytree(web,incoming/(generation+'-frontend'))
        write(incoming/(generation+'-backend'),python+(b'\nOS-OTA-D09-generation-two\n' if generation=='new' else b''),0o755)
    write(incoming/'new-frontend/index.html','new fixture frontend; actual OS worker test\n')
    config=(PACKAGE/'image-recipe/configs/60-terminal-workspace.conf').read_bytes()
    write(incoming/'new-config',config)
    old_effective=wire.digest(owner.effective_configuration().encode())
    write(owner.DESTINATIONS['config'],config)
    run('systemctl','daemon-reload')
    new_effective=wire.digest(owner.effective_configuration().encode())
    owner.DESTINATIONS['config'].unlink()
    run('systemctl','daemon-reload')
    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=secrets.token_hex(16),
        grants={r:secrets.token_hex(32) for r in ('UPDATER','AUTOMATIC','MANUAL','GUARD','OBSERVER')},
        objects={name:dict(old=owner.object_digest(destination),new=owner.object_digest(incoming/('new-'+name)),
                          restore=owner.object_digest(incoming/('restore-'+name))) for name,destination in owner.DESTINATIONS.items()},
        artifacts=dict(backend=wire.digest((incoming/'new-backend').read_bytes()),frontend=owner.object_digest(incoming/'new-frontend')),
        restore_artifacts=dict(backend=wire.digest(python),frontend=owner.object_digest(web)),
        effective_config=dict(APPLY=new_effective,RESTORE=old_effective),
        exclusion_evidence_digest=wire.digest(exclusion),helper_digest=wire.digest(Path('/usr/libexec/archipelago-d09-prepare-workspace').read_bytes()),
        prerequisite=dict(grant_digest=secrets.token_hex(32),backend_sha256=wire.digest(python),effective_config_digest=old_effective))
    return provision,binding


def request(args,stdin=''):
    key=secrets.token_hex(8)
    write(CONTROL/('request-'+key+'.json'),json.dumps(dict(args=args,stdin=stdin)))
    output=CONTROL/('result-'+key+'.json')
    wait(output.exists,'client reply',40)
    value=json.loads(output.read_text())
    print('ACTUAL_UPDATER_CLIENT',json.dumps(value),flush=True)
    return value


def execute(context,os_binding,plan,scenario):
    import os_update_executor as engine
    import d09_wire as wire
    import d09_store as store
    provision,binding=context
    owner=provision.owner
    binding['os_update']=os_binding['os_update']
    version='os-ota-disposable-20261103'
    binding['release_identity']=[version,sorted([
        ['archipelago',version,binding['artifacts']['backend'],(owner.ROOT/'incoming/new-backend').stat().st_size],
        ['os-update-plan.json',version,binding['os_update']['plan_sha256'],len(wire.canonical(plan))]])]
    source=engine.ROOT/'plans'/binding['os_update']['plan_sha256']
    os_inputs=owner.ROOT/'incoming/os-update'
    os_inputs.mkdir(mode=0o700)
    for name in ('plan.json','manifest.json','preservation.json'):
        shutil.copyfile(source/name,os_inputs/name)
    shutil.copyfile(engine.VERIFIER,os_inputs/'verifier')
    admission=owner.ROOT/'incoming/admission.json'
    store.atomic(admission,wire.canonical(binding))
    provision.provision(admission)
    run('systemctl','start','archipelago-d09-lifecycle.service')
    wait(lambda:owner.PRIVATE.exists() and owner.SOCKET.exists(),'live owner sockets',30)
    wait(lambda:provision.activate_prerequisite()['prerequisite_started'],'one-use prerequisite start',60)
    prior=owner.service()['InvocationID']
    inspect=request(['inspect'])
    assert inspect['exit']==0,inspect
    observed=json.loads(inspect['stdout'])
    assert observed['os_update']['supported'] and observed['os_update']['plan_sha256']==binding['os_update']['plan_sha256']
    (EVIDENCE/'d09-binding.json').write_bytes(wire.canonical(binding))
    before=engine.inventory()
    canaries=Path('/var/lib/archipelago/ota-synthetic-canaries')
    hashes=lambda:{p.name:engine.file_hash(p) for p in canaries.iterdir()}
    preserved=hashes()
    lock=None
    result=dict(production_approved=False,native_application_qualified=False,actual_d09_client_owner_worker=True,scenario=scenario,passed=False)
    try:
        if scenario=='d09-locked-apt':
            lock=open('/var/lib/dpkg/lock-frontend','a')
            fcntl.lockf(lock,fcntl.LOCK_EX|fcntl.LOCK_NB)
        response=request(['prepare','--selected-release',json.dumps(binding['release_identity']),
                          '--staging-generation',secrets.token_hex(16)])
        assert response['exit']==0,response
        def terminal():
            value=json.loads((owner.ROOT/'ledger.json').read_text())
            receipt=value.get('receipt') or {}
            return value if receipt.get('phase') in ('CONFLICT','AWAITING_OBSERVATION','COMPLETE') else None
        ledger=wait(terminal,'actual D09 worker terminal phase')
        status=engine.Executor(binding['os_update']).status
        result.update(d09_phase=ledger['receipt']['phase'],os_phase=status['phase'],
                      prior_invocation=prior,observed_invocation=owner.service()['InvocationID'])
        logs='\n'.join(p.read_text() for p in engine.Executor(binding['os_update']).root.glob('apt-*.log'))
        (EVIDENCE/'d09-apt.log').write_text(logs)
        if scenario in ('d09-locked-apt','d09-missing-dependency'):
            assert ledger['receipt']['phase']=='CONFLICT'
            assert owner.service()['InvocationID']==prior and owner.service()['ActiveState']=='active'
            assert engine.inventory()==before
            if scenario=='d09-locked-apt':
                assert 'lock-frontend' in logs and ('Could not' in logs or 'Unable' in logs),logs
            else:
                assert status['phase']=='DOWNLOAD'
                # Preserve the actual executor exception from the real worker.
                unit=ledger['execution']['unit']
                journal=run('journalctl','-u',unit,'--no-pager').stdout
                (EVIDENCE/'d09-worker.log').write_text(journal)
                assert 'incomplete archive transaction' in journal,journal
            result['manager_uninterrupted']=True
        else:
            assert ledger['receipt']['phase']=='AWAITING_OBSERVATION',ledger['receipt']
            assert status['phase']=='REBOOT_REQUIRED'
            assert owner.service()['InvocationID']!=prior
            assert hashes()==preserved
            with (canaries/'wallet.db').open('a') as stream:
                stream.write('FORWARD-AFTER-D09-OS-APPLY-sequence-18\n')
            (EVIDENCE/'canaries-after.json').write_bytes(wire.canonical(hashes()))
            assert engine.stage(binding,'APPLY','apply')==status
            accepted=next(v['request'] for v in ledger['requests'].values() if v['request']['operation']=='APPLY')
            original=(owner.ROOT/'ledger.json').read_bytes()
            replay=request(['submit'],wire.canonical(accepted).decode())
            assert replay['exit']==0 and json.loads(replay['stdout'])['status']=='REFUSED',replay
            assert (owner.ROOT/'ledger.json').read_bytes()==original
            grub=Path('/boot/grub/grub.cfg').read_text()
            assert plan['boot']['running_release'] in grub and plan['boot']['expected_release'] in grub
            result.update(replay_refused=True,os_repeat_idempotent=True,retained_old_grub_entry=True,
                          forward_wallet_write_preserved=True,boot_id=engine.boot_id(),target_kernel=plan['boot']['expected_release'])
            (EVIDENCE/'apply-result.json').write_bytes(wire.canonical(result))
        result.update(passed=True,canaries_preserved=hashes()==preserved if scenario!='d09-apply' else True)
    finally:
        if lock:
            lock.close()
        (EVIDENCE/'d09-result.json').write_text(json.dumps(result,indent=2)+'\n')
        run('journalctl','-u','archipelago-d09-lifecycle.service','--no-pager',check=False)
        print('D09_INSTALLED_RESULT',json.dumps(result),flush=True)