#!/usr/bin/python3
"""Disposable PID1 kiosk/audio exercise. Hardware seams are explicit.

ALSA device IO is replaced by regular files with REAL Linux ACLs. Xorg uses its
dummy video driver. PipeWire/Pulse/WirePlumber and Chromium are real processes.
No host device, profile, service bus, wallet or runtime graphroot is mounted.
"""
import json
import hashlib
import os
from pathlib import Path
import shutil
import socket
import stat
import subprocess
import sys
import time

ROOT = Path('/opt/kiosk-fixture')
sys.path.insert(0, str(ROOT/'scripts'))


def command(*args, check=True):
    result = subprocess.run(args,capture_output=True,text=True,timeout=90)
    if check and result.returncode:
        raise RuntimeError((args,result.returncode,result.stdout,result.stderr))
    return result.stdout


def wait(predicate, message, seconds=40):
    end=time.monotonic()+seconds
    while time.monotonic()<end:
        if predicate():return
        time.sleep(.25)
    raise AssertionError(message)


def fixture_guard():
    assert os.geteuid()==0 and Path('/run/.containerenv').exists()
    assert socket.gethostname()=='c01-kiosk-fixture'
    assert Path('/proc/1/comm').read_text().strip()=='systemd'


def audio_daemon():
    import c01_audio_devices as audio
    class FakeALSA(audio.Device):
        def __init__(self,item):
            self.item=item;self.path=Path('/run/fixture-sound')/item['name']
            self.fd=os.open(self.path,os.O_PATH|os.O_NOFOLLOW|os.O_CLOEXEC)
            self.handle=Path('/proc/self/fd')/str(self.fd)
            self.original_identity=audio.identity(os.fstat(self.fd))
            self.check_kernel()
        def check_kernel(self):
            meta=self.path.lstat()
            audio.require(stat.S_ISREG(meta.st_mode) and audio.identity(meta)==self.original_identity and
                          meta.st_uid==0 and meta.st_gid==29,'fixture ALSA incarnation replaced')
    audio.Device=FakeALSA
    audio.device_path=lambda item:Path('/run/fixture-sound')/item['name']
    sys.argv=['c01_audio_devices.py',sys.argv[2]]
    audio.main()


def put(path,text,mode=0o644):
    path=Path(path);path.parent.mkdir(parents=True,exist_ok=True)
    path.write_text(text);path.chmod(mode)


def user(*args,check=True):
    return command('runuser','-u','archipelago','--','env','XDG_RUNTIME_DIR=/run/user/1000',
                   'DBUS_SESSION_BUS_ADDRESS=unix:path=/run/user/1000/bus',*args,check=check)


def main():
    fixture_guard()
    if len(sys.argv)>1 and sys.argv[1]=='audio-owner':
        audio_daemon();return
    import c01_audio_devices as audio
    command('systemctl','stop','user@1000.service',check=False)
    command('usermod','-G','','-s','/usr/sbin/nologin','archipelago')
    # Original operator-home test secrets must be hidden by the runtime view.
    put('/home/archipelago/.ssh/fixture-secret','DO-NOT-EXPOSE\n',0o600)
    for name in ('pulse','wireplumber'):
        path=Path('/home/archipelago/.config/pulse' if name=='pulse' else '/home/archipelago/.local/state/wireplumber')
        path.mkdir(parents=True,exist_ok=True);os.chown(path,1000,1000)
    cookie=Path('/home/archipelago/.config/pulse/cookie')
    cookie.write_bytes(b'F'*256);cookie.chmod(0o600);os.chown(cookie,1000,1000)
    cookie_identity=(cookie.stat().st_dev,cookie.stat().st_ino,hashlib.sha256(cookie.read_bytes()).hexdigest())
    command('install','-d','-o','root','-g','root','-m','0700','/var/lib/fixture-runtime')
    command('install','-d','-o','archipelago','-g','archipelago','-m','0700','/var/lib/fixture-runtime/home')
    for name in ('.config','.cache','.local','.local/state','.config/pulse','.local/state/wireplumber'):
        command('install','-d','-o','archipelago','-g','archipelago','-m','0700','/var/lib/fixture-runtime/home/'+name)
    for name in ('manager','snapshot','authority'):
        command('install','-d','-m','0700','/var/lib/fixture-'+name)
    command('install','-d','-m','0700',str(audio.STATE))
    command('install','-d','-m','0755','/run/fixture-sound','/tmp/.X11-unix','/dev/snd')
    os.chmod('/tmp/.X11-unix',0o1777)
    device=Path('/run/fixture-sound/controlC0');device.touch();os.chown(device,0,29);device.chmod(0o660)
    policy=dict(version=1,uid=1000,prior_groups=[20,27,29,1000],devices=[dict(name='controlC0',
        sysfs='/sys/devices/fixture/sound/card0/controlC0',rdev=os.makedev(116,0),uid=0,gid=29,mode=0o660,acl_sha256=None)])
    put(audio.CONFIG,json.dumps(policy),0o600)
    configs=ROOT/'image-recipe/configs'
    for src,dest in (('archipelago-kiosk-launcher.sh','/usr/local/bin/archipelago-kiosk-launcher'),
                     ('archipelago-audio-router.sh','/usr/local/bin/archipelago-audio-router')):
        put(dest,(configs/src).read_text(),0o755)
    put('/usr/lib/archipelago/c01_audio_devices.py',(ROOT/'scripts/c01_audio_devices.py').read_text(),0o644)
    put('/etc/systemd/user/fixture-playback.service','''[Service]
ExecStart=/usr/bin/paplay --device=fixture_audio --raw --format=s16le --rate=48000 --channels=2 /dev/zero
Restart=no
''')
    put('/etc/systemd/user/fixture-audio-reader.service','''[Service]
ExecStart=/usr/bin/python3 -c "import time; f=open('/run/fixture-sound/controlC0'); time.sleep(300)"
Restart=no
''')
    for name in ('archipelago-kiosk-browser.service','archipelago-runtime-audio-router.service','archipelago-runtime-nostr-relay.service'):
        put('/etc/systemd/user/'+name,(configs/name).read_text())
    for name in ('c01_network','c01_endpoints','runtime_uplink','c01_uplink','c01_network_mdns','c01_retained_producer','c01_network_plan','c01_network_owner','c01_kiosk_endpoints'):
        put('/usr/lib/archipelago/'+name+'.py',(ROOT/'scripts'/f'{name}.py').read_text())
    for name in ('archipelago-c01-network.service','archipelago-c01-kiosk-endpoints.service'):
        put('/etc/systemd/system/'+name,(configs/name).read_text())
    command('install','-d','-o','archipelago','-g','archipelago','-m','0700','/var/lib/archipelago/nostr-relay')
    put('/var/lib/archipelago/nostr-relay/config.toml',(configs/'nostr-relay-config.toml').read_text())
    put('/var/lib/archipelago/nostr-relay/preservation-marker','retained-relay\n')
    # Real systemd user manager; exact existing private home/X11/state templates.
    drop=(configs/'80-archipelago-runtime-user.conf.in').read_text()
    for key,value in {'ORIGINAL_RUNTIME_UID':'1000','ORIGINAL_HOME':'/home/archipelago',
        'SANITIZED_RUNTIME_HOME':'/var/lib/fixture-runtime/home','MANAGER_BACKING_PARENT':'/var/lib/fixture-manager',
        'SNAPSHOT_BACKING_PARENT':'/var/lib/fixture-snapshot','AUTHORITY_BACKING_PARENT':'/var/lib/fixture-authority'}.items():
        drop=drop.replace('@'+key+'@',value)
    put('/etc/systemd/system/user@1000.service.d/80-runtime.conf',drop)
    for name in ('84-c01-kiosk-x11.conf','86-c01-runtime-audio-state.conf','87-c01-runtime-audio-owner.conf'):
        put('/etc/systemd/system/user@1000.service.d/'+name,(configs/name).read_text())
    command('ip','netns','add','archipelago-runtime')
    command('ip','-n','archipelago-runtime','link','set','lo','up')
    command('ip','netns','add','archipelago-managed')
    command('ip','-n','archipelago-managed','link','set','lo','up')
    put('/run/fixture-web/index.html','<html><body>C01 Kiosk fixture</body></html>')
    put('/etc/nginx/nginx.conf','''worker_processes 1;
events { worker_connections 128; }
http { server { listen 80; location / { root /run/fixture-web; try_files $uri /index.html; } } }
''')
    put('/etc/systemd/system/nginx.service','[Service]\nType=simple\nExecStart=/usr/sbin/nginx -g "daemon off;"\n')
    import c01_network_owner as network
    runtime=dict(runtime_uid=1000,runtime_gid=1000,runtime_user='archipelago',user_cgroup='/user.slice/user-1000.slice/user@1000.service')
    put('/etc/archipelago-c01/runtime.json',json.dumps(runtime),0o600)
    manifests=[];routes=[dict(name='kiosk-http',source='runtime',target='host',protocol='tcp',port=80,
        app_id=None,projection_sha256=None,target_uid=0,target_cgroup='/system.slice/nginx.service')]
    for app,port,unit,source in (('archipelago-kiosk',9222,'archipelago-kiosk-browser.service','managed'),
                                ('nostr-relay',7777,'archipelago-runtime-nostr-relay.service','host')):
        group=runtime['user_cgroup']+'/app.slice/'+unit
        doc=dict(app_id=app,manifest_sha256=hashlib.sha256((configs/unit).read_bytes()).hexdigest(),
                 loopback_ports=[['tcp',port]],target_cgroup=group)
        manifests.append(doc)
        routes.append(dict(name=f'app/{app}/tcp/{port}/{source}',source=source,target='runtime',protocol='tcp',port=port,
            app_id=app,projection_sha256=network.digest(network.canonical(doc)),target_uid=1000,target_cgroup=group))
    namespaces={}
    for role,path in (('host','/proc/1/ns/net'),('runtime','/run/netns/archipelago-runtime'),('managed','/run/netns/archipelago-managed')):
        m=os.stat(path);namespaces[role]=[m.st_dev,m.st_ino]
    endpoint_raw=network.canonical(dict(version=1,namespaces=namespaces,manifests=manifests,routes=routes))
    put('/etc/archipelago-c01/endpoints.json',endpoint_raw.decode(),0o600)
    executables={p:hashlib.sha256(Path(p).read_bytes()).hexdigest() for p in ('/usr/lib/chromium/chromium','/usr/local/bin/nostr-rs-relay')}
    put('/etc/archipelago-c01/kiosk-endpoints.json',json.dumps(dict(version=1,policy_sha256=network.digest(endpoint_raw),
        executables=executables,routes=[r['name'] for r in routes if r['target']=='runtime'])),0o600)
    command('install','-d','/opt/archipelago')
    access=(configs/'81-c01-runtime-access.conf.in').read_text()
    for key,value in {'MANAGER_PARENT':'/var/lib/fixture-manager','SNAPSHOT_PARENT':'/var/lib/fixture-snapshot',
                      'AUTHORITY_PARENT':'/var/lib/fixture-authority'}.items():access=access.replace('@'+key+'@',value)
    # Egress is outside this targeted loopback/hardware fixture. Retain EVERY
    # production view/exec/IPC restriction, overriding only the external uplink
    # readiness probe already qualified in the separate runtime composition.
    put('/etc/systemd/system/user@1000.service.d/81-private.conf',access)
    put('/etc/systemd/system/user@1000.service.d/99-fixture-uplink.conf','[Service]\nExecCondition=\nExecCondition=/usr/bin/true\n')
    owner=(configs/'archipelago-audio-devices.service').read_text().replace(
        '/usr/bin/python3 -I -S /usr/lib/archipelago/c01_audio_devices.py',
        '/usr/bin/python3 -B /opt/kiosk-fixture/scripts/tests/c01_kiosk_pid1.py audio-owner')
    owner=owner.replace('ReadWritePaths=/dev/snd ','ReadWritePaths=/run/fixture-sound ')
    put('/etc/systemd/system/archipelago-audio-devices.service',owner)
    recovery=(configs/'archipelago-audio-reconcile.service').read_text().replace(
        '/usr/bin/python3 -I -S /usr/lib/archipelago/c01_audio_devices.py',
        '/usr/bin/python3 -B /opt/kiosk-fixture/scripts/tests/c01_kiosk_pid1.py audio-owner').replace(
        'ReadWritePaths=/dev/snd ','ReadWritePaths=/run/fixture-sound ')
    put('/etc/systemd/system/archipelago-audio-reconcile.service',recovery)
    put('/etc/X11/fixture-dummy.conf','''Section "ServerFlags"
 Option "AutoAddDevices" "false"
 Option "AutoAddGPU" "false"
EndSection
Section "Device"
 Identifier "dummy"
 Driver "dummy"
 VideoRam 256000
EndSection
Section "Monitor"
 Identifier "monitor"
 HorizSync 5.0-1000.0
 VertRefresh 5.0-200.0
 Modeline "1280x720" 74.50 1280 1344 1472 1664 720 723 728 748
EndSection
Section "Screen"
 Identifier "screen"
 Device "dummy"
 Monitor "monitor"
 DefaultDepth 24
 SubSection "Display"
  Depth 24
  Modes "1280x720"
 EndSubSection
EndSection
''')
    # Run the complete production root launcher. Its Xorg executable appointment
    # is the explicit hardware seam: real dummy-driver Xorg, without physical VT.
    Path('/usr/bin/Xorg').unlink()
    put('/usr/bin/Xorg','#!/bin/sh\nexec /usr/lib/xorg/Xorg :0 -config /etc/X11/fixture-dummy.conf -noreset -nolisten tcp\n',0o755)
    put('/etc/systemd/system/archipelago-kiosk.service','''[Service]
Environment=ARCHIPELAGO_C01_KIOSK=1
ExecStart=/usr/local/bin/archipelago-kiosk-launcher
Restart=no
''')
    command('systemctl','daemon-reload')
    command('systemctl','start','nginx.service','archipelago-c01-network.service')
    command('systemctl','start','archipelago-kiosk.service')
    wait(lambda:Path('/tmp/.X11-unix/X0').exists(),'dummy Xorg socket')
    wait(lambda:Path('/run/archipelago-kiosk/browser.env').exists(),'production display environment handoff')
    command('install','-d','-o','archipelago','-g','archipelago','-m','0700','/var/lib/archipelago/chromium-kiosk')
    put('/var/lib/archipelago/chromium-kiosk/preservation-marker','retained-profile\n')
    original=Path('/var/lib/archipelago/chromium-kiosk').stat().st_ino
    command('systemctl','start','user@1000.service')
    assert audio.permissions(audio.decode(audio.acl(device),device.stat().st_mode),0,29,1000,[1000])==6
    user('systemctl','--user','start','pipewire.service','pipewire-pulse.service','wireplumber.service')
    user('pactl','load-module','module-null-sink','sink_name=fixture_audio','sink_properties=device.description=FixtureAudio')
    def sink_ready():
        try:user('pactl','set-default-sink','fixture_audio');return True
        except RuntimeError:return False
    wait(sink_ready,'WirePlumber default-sink metadata')
    user('systemctl','--user','start','fixture-playback.service')
    wait(lambda:'RUNNING' in user('pactl','list','short','sinks'),'real null-sink playback running')
    # Let the fresh null node's WirePlumber restore job finish before setting
    # the synthetic *prior* operator state tested across the kiosk startup.
    time.sleep(3)
    user('pactl','set-sink-volume','fixture_audio','37%')
    user('pactl','set-sink-mute','fixture_audio','1')
    wait(lambda:user('pactl','get-sink-mute','fixture_audio').strip()=='Mute: yes' and
         '37%' in user('pactl','get-sink-volume','fixture_audio'),'synthetic prior sink state applied')
    print(json.dumps({'before_browser':{'default':user('pactl','get-default-sink').strip(),
        'mute':user('pactl','get-sink-mute','fixture_audio').strip(),
        'volume':user('pactl','get-sink-volume','fixture_audio').strip()}}),flush=True)
    assert user('pactl','get-sink-mute','fixture_audio').strip()=='Mute: yes','synthetic prior mute not established'
    user('systemctl','--user','start','archipelago-runtime-audio-router.service')
    user('systemctl','--user','start','archipelago-kiosk-browser.service')
    user('systemctl','--user','start','archipelago-runtime-nostr-relay.service')
    command('systemctl','start','archipelago-c01-kiosk-endpoints.service')
    wait(lambda:bool(user('systemctl','--user','show','archipelago-kiosk-browser.service','-p','MainPID','--value').strip() not in ('','0')),'browser service')
    # CDP must be a real listening Chromium socket in the private runtime realm.
    wait(lambda:'9222' in command('ip','netns','exec','archipelago-runtime','ss','-ltn',check=False),'real Chromium CDP',60)
    state=json.loads(command('ip','netns','exec','archipelago-runtime','curl','-fsS','http://127.0.0.1:9222/json/version'))
    assert 'Chrome' in state['Browser']
    def bridged_cdp(previous=None):
        try:
            response=json.loads(command('ip','netns','exec','archipelago-managed','curl','-fsS','--max-time','2','http://127.0.0.1:9222/json/version'))
            return 'Chrome' in response['Browser'] and response['webSocketDebuggerUrl']!=previous
        except (RuntimeError,ValueError,KeyError):return False
    wait(bridged_cdp,'real CDP through pinned managed-to-runtime bridge')
    assert 'C01 Kiosk fixture' in command('ip','netns','exec','archipelago-runtime','curl','-fsS','http://127.0.0.1/kiosk')
    def relay_ready():
        try:return bool(command('curl','-fsS','--max-time','2','-H','Accept: application/nostr+json','http://127.0.0.1:7777/'))
        except RuntimeError:return False
    wait(relay_ready,'real relay through host-to-runtime bridge')
    first_browser=state['webSocketDebuggerUrl']
    user('systemctl','--user','restart','archipelago-kiosk-browser.service')
    wait(bridged_cdp,'CDP producer re-registration after browser restart')
    state=json.loads(command('ip','netns','exec','archipelago-managed','curl','-fsS','http://127.0.0.1:9222/json/version'))
    assert state['webSocketDebuggerUrl']!=first_browser
    before_display_restart=state['webSocketDebuggerUrl']
    command('systemctl','restart','archipelago-kiosk.service')
    wait(lambda:bridged_cdp(before_display_restart),'CDP re-admission after root display restart')
    state=json.loads(command('ip','netns','exec','archipelago-managed','curl','-fsS','http://127.0.0.1:9222/json/version'))
    assert state['webSocketDebuggerUrl']!=before_display_restart
    pid=int(command('systemctl','show','user@1000.service','-p','MainPID','--value').strip())
    assert not Path(f'/proc/{pid}/root/home/archipelago/.ssh').exists()
    assert not Path(f'/proc/{pid}/root/home/archipelago/.pki').exists(),'browser created forbidden top-level operator credential directory'
    assert not Path(f'/proc/{pid}/root/home/archipelago/.gnupg').exists()
    for name in ('/run/dbus/system_bus_socket','/run/systemd/private','/run/sudo','/run/polkit-1',
                 '/usr/bin/sudo','/usr/bin/pkexec','/usr/bin/su','/usr/bin/doas','/usr/bin/chsh',
                 '/usr/bin/chfn','/usr/bin/passwd','/usr/bin/crontab','/usr/bin/at','/root',
                 '/home/archipelago-admin','/var/lib/fixture-manager','/var/lib/fixture-snapshot','/var/lib/fixture-authority'):
        try:meta=Path(f'/proc/{pid}/root'+name).stat()
        except FileNotFoundError:continue
        assert meta.st_uid==0 and stat.S_IMODE(meta.st_mode)==0,('privileged view exposed',name)
    for name in ('/opt/archipelago','/usr/local/bin'):
        assert os.statvfs(f'/proc/{pid}/root'+name).f_flag & os.ST_RDONLY,('mutable package view',name)
    assert Path(f'/proc/{pid}/root/home/archipelago').stat().st_ino==Path('/var/lib/fixture-runtime/home').stat().st_ino
    assert Path('/var/lib/archipelago/chromium-kiosk').stat().st_ino==original
    assert (cookie.stat().st_dev,cookie.stat().st_ino,hashlib.sha256(cookie.read_bytes()).hexdigest())==cookie_identity
    assert user('pactl','get-default-sink').strip()=='fixture_audio'
    mute=user('pactl','get-sink-mute','fixture_audio').strip()
    assert mute=='Mute: yes',mute
    assert '37%' in user('pactl','get-sink-volume','fixture_audio')
    user('systemctl','--user','restart','archipelago-runtime-audio-router.service')
    time.sleep(6)
    assert user('pactl','get-default-sink').strip()=='fixture_audio'
    assert user('pactl','get-sink-mute','fixture_audio').strip()=='Mute: yes'
    assert '37%' in user('pactl','get-sink-volume','fixture_audio')
    # Owner stop must retire the complete runtime BEFORE preimage restoration.
    command('systemctl','stop','archipelago-audio-devices.service')
    wait(lambda:command('systemctl','is-active','user@1000.service',check=False).strip()=='inactive','runtime dependent retirement')
    audio.runtime_retired()
    assert audio.acl(device) is None and stat.S_IMODE(device.stat().st_mode)==0o660
    assert all(json.loads(p.read_text())['phase']=='revoked' for p in audio.STATE.glob('*.json'))
    command('systemctl','start','user@1000.service')
    user('systemctl','--user','start','fixture-audio-reader.service')
    reader=int(user('systemctl','--user','show','fixture-audio-reader.service','-p','MainPID','--value').strip())
    assert reader>1
    command('systemctl','kill','--kill-whom=main','--signal=KILL','archipelago-audio-devices.service')
    wait(lambda:command('systemctl','is-active','user@1000.service',check=False).strip()=='inactive','runtime retirement after owner SIGKILL')
    assert not Path(f'/proc/{reader}').exists(),'open-device consumer survived revocation'
    wait(lambda:audio.acl(device) is None,'ACL restoration after owner SIGKILL')
    command('systemctl','reset-failed','archipelago-audio-devices.service')
    command('systemctl','start','user@1000.service')
    user('systemctl','--user','start','fixture-audio-reader.service')
    device.unlink();device.touch();os.chown(device,0,29);device.chmod(0o660)
    wait(lambda:command('systemctl','is-active','user@1000.service',check=False).strip()=='inactive','runtime retirement on device incarnation replacement')
    assert audio.acl(device) is None,'old grant followed a replacement node'
    wait(lambda:all(json.loads(p.read_text())['phase'] in ('revoked','detached') for p in audio.STATE.glob('*.json')),
         'detached incarnation journal reconciliation')
    command('systemctl','reset-failed','archipelago-audio-devices.service')
    command('systemctl','start','user@1000.service')
    assert audio.permissions(audio.decode(audio.acl(device),device.stat().st_mode),0,29,1000,[1000])==6
    command('systemctl','stop','archipelago-audio-devices.service')
    assert audio.acl(device) is None,'fresh incarnation preimage not restored'
    print(json.dumps({'real_pid1':True,'real_xorg_dummy':True,'real_chromium_cdp':state['Browser'],
        'real_pipewire_null_sink':True,'real_acl_roundtrip':True,'runtime_retired_before_revoke':True,
        'real_relay':True,'real_pinned_http_cdp_relay_routes':True,'cdp_restart_reregistered':True,
        'owner_sigkill_retired_open_device_consumer':True,'device_replacement_blocked_without_regrant':True,
        'fresh_device_readmission_after_retirement':True,
        'profile_inode_preserved':True,'original_pulse_cookie_inode_bytes_preserved':True,
        'physical_audio_accepted':False,'full_native_cutover_accepted':False}),flush=True)


if __name__=='__main__':main()