#!/usr/bin/python3
"""Real mount/UID/process recovery test in an explicitly private namespace."""
import ctypes
import hashlib
import json
import os
from pathlib import Path
import subprocess
import sys

ROOT=Path(__file__).resolve().parents[2]
sys.path.insert(0,str(ROOT/'scripts'))
import os_update_mount_custody as coordinator
from d09_store import atomic

assert os.geteuid()==0 and os.readlink('/proc/self/ns/mnt')!=sys.argv[1]
libc=ctypes.CDLL(None,use_errno=True)
def mount_tmp(path):
    assert libc.mount(b'tmpfs',os.fsencode(path),b'tmpfs',2|4,b'size=64m,mode=0755')==0
mount_tmp('/var/lib');mount_tmp('/etc')
data=Path('/var/lib/archipelago');data.mkdir();mount_tmp(data);os.chown(data,1000,1000)
support=data/'support';support.mkdir(mode=0o755)
old=support/'previous-evidence';old.write_bytes(b'retained old evidence\n')
oldmeta=old.stat();before=data.stat()
fd=os.open(support,os.O_PATH|os.O_DIRECTORY)
expected=coordinator.identity(fd);os.close(fd)
value={'version':1,'node':os.uname().nodename,'source':str(support),'source_identity':expected,
       'alias':'/var/lib/archipelago-delivery-support','os_root':'/var/lib/archipelago-os-update',
       'stage_name':'delivery-os-custody'}
atomic(coordinator.CONFIG,json.dumps(value,sort_keys=True).encode())
entry=str(ROOT/'scripts/os_update_mount_custody.py')
def recover():
    result=subprocess.run(['/usr/bin/python3','-B',entry,'recover'],capture_output=True,text=True,timeout=20)
    assert result.returncode==0,result.stderr
    return json.loads(result.stdout)
first=recover()
root=Path(value['os_root']);sentinel=root/'journal-sentinel'
sentinel.write_bytes(b'never reset existing updater journal\n');sentinel.chmod(0o600)
retained=(sentinel.stat().st_dev,sentinel.stat().st_ino,hashlib.sha256(sentinel.read_bytes()).hexdigest())
second=recover();assert second['stage_identity']==first['stage_identity'] and second['sequence']>first['sequence']
# Simulate loss of volatile mount state while retaining both filesystems and
# durable OS coordinator journal. Recovery runs in a new process, not a reused FD.
assert libc.umount2(os.fsencode(root),0)==0
assert libc.umount2(os.fsencode(value['alias']),0)==0
third=recover();assert third['stage_identity']==first['stage_identity']
assert (sentinel.stat().st_dev,sentinel.stat().st_ino,hashlib.sha256(sentinel.read_bytes()).hexdigest())==retained
assert (data.stat().st_dev,data.stat().st_ino,data.stat().st_uid,data.stat().st_gid)==(before.st_dev,before.st_ino,1000,1000)
pid=os.fork()
if pid==0:
    try:
        os.setgroups([]);os.setgid(1000);os.setuid(1000)
        (data/'runtime-write-control').write_text('allowed')
        for path in (root/'hostile',Path(value['alias'])/'hostile',coordinator.STATE/'hostile'):
            try:path.write_text('forbidden')
            except PermissionError:pass
            else:os._exit(10)
        support.rename(data/'support-retained');support.mkdir();(support/'hostile').write_text('untrusted')
        os._exit(0)
    except BaseException:os._exit(11)
assert os.waitpid(pid,0)[1]==0,'real UID1000 negative failed'
failed=subprocess.run(['/usr/bin/python3','-B',entry,'recover'],capture_output=True,text=True,timeout=20)
assert failed.returncode!=0 and 'original support inode/custody differs' in failed.stderr
assert not (support/'delivery-os-custody').exists()
assert (Path(value['alias'])/'previous-evidence').stat().st_ino==oldmeta.st_ino
assert (Path(value['alias'])/'previous-evidence').read_bytes()==b'retained old evidence\n'
assert (sentinel.stat().st_dev,sentinel.stat().st_ino,hashlib.sha256(sentinel.read_bytes()).hexdigest())==retained
print(json.dumps({'passed':True,'first':first,'recovered':third,'actual_uid1000_attack_refused':True,
 'source_path_swap_refused':True,'existing_evidence_preserved':True,'data_root_ownership_preserved':True,
 'new_process_mount_recovery':True,'existing_OS_journal_preserved':True,'physical_power_operation':False,
 'scope':'private mount namespace; durable process/mount recovery, not installed-node acceptance'},sort_keys=True))