#!/usr/bin/python3
"""Normal native startup boundary in an owned rootless PID1 realm.

Exit 77 is BLOCKED, not a readiness pass. The prerequisite driver is retained as
a separate historical sub-exit. No grants, leases, APP_READY or observer results
are fabricated. Run only with test-c01-normal-systemd-isolated.sh.
"""
import hashlib
import json
import os
from pathlib import Path
import socket
import subprocess
import time
import threading

PACKAGE = Path('/opt/c01-fixture')
EVIDENCE = Path('/var/lib/c01-normal')


def run(*args, timeout=120):
    return subprocess.run(args, capture_output=True, text=True, timeout=timeout)


def record(name, value):
    (EVIDENCE / name).write_text(json.dumps(value, indent=2) + '\n')


def native_boundary(selector, label, expected, slots=(), passcred=False):
    # An actual ordinary executable; intentionally no libtest entry selector.
    command = ['/usr/local/bin/archipelago', selector]
    channels = socket.socketpair(socket.AF_UNIX, socket.SOCK_SEQPACKET) if slots else ()
    if channels and passcred:
        channels[0].setsockopt(socket.SOL_SOCKET, socket.SO_PASSCRED, 1)
    def setup():
        for slot in slots:
            os.dup2(channels[0].fileno(), slot, inheritable=True)
    try:
        p = subprocess.run(command, capture_output=True, text=True, timeout=30,
                           close_fds=False, preexec_fn=setup)
    finally:
        for channel in channels:
            channel.close()
    record(label + '.json', {'command': command, 'exit': p.returncode,
        'stdout': p.stdout, 'stderr': p.stderr, 'foreign_slots': slots, 'passcred': passcred,
        'expected_boundary': expected})
    if p.returncode == 0:
        raise AssertionError('unexpected native capability-free acceptance: ' + label)
    assert expected in p.stderr, (label, p.stderr)
    print('NATIVE REFUSAL', label, p.returncode, p.stderr.strip(), flush=True)


def main():
    assert os.geteuid() == 0 and Path('/run/.containerenv').exists()
    assert os.environ.get('C01_NORMAL_PID1') == '1'
    assert socket.gethostname() == 'c01-prerequisite-qualification'
    assert Path('/proc/1/comm').read_text().strip() == 'systemd'
    EVIDENCE.mkdir(mode=0o700)
    manifest = json.loads((PACKAGE / 'source-sha256.json').read_text())
    for name, digest in manifest.items():
        assert hashlib.sha256((PACKAGE / name).read_bytes()).hexdigest() == digest, name
    record('source-verified.json', {'files': len(manifest), 'manifest_sha256':
        hashlib.sha256((PACKAGE / 'source-sha256.json').read_bytes()).hexdigest()})
    archive=PACKAGE/('normal-native.tar.gz' if (PACKAGE/'normal-native.tar.gz').exists() else 'normal-native.tar')
    preinstalled=PACKAGE/'native-preinstalled.json'
    stack={};derivation=[]
    if preinstalled.exists():
        proof=json.loads(preinstalled.read_text())
        with archive.open('rb') as stream:assert hashlib.file_digest(stream,'sha256').hexdigest()==proof['archive_sha256']
        if os.environ.get('C01_RUNNING_SERVICES')=='1':
            with (PACKAGE/'service-stack.tar.gz').open('rb') as stream:
                assert hashlib.file_digest(stream,'sha256').hexdigest()==proof['service_stack_sha256']
            stack=json.loads((PACKAGE/'service-stack-manifest.json').read_text())['files']
            assert 'usr/local/bin/archipelago' not in stack
            for name,item in stack.items():assert hashlib.sha256(Path('/'+name).read_bytes()).hexdigest()==item['sha256']
    else:
        assert os.environ.get('C01_RUNNING_SERVICES')!='1','running profile dependencies must be installed before PID1'
        installed = run('tar', '-xf', str(archive), '-C', '/')
        assert installed.returncode == 0, installed.stderr
    for name, item in json.loads((PACKAGE / 'native-artifact-sha256.json').read_text()).items():
        canonical='usr/'+name if name.startswith(('lib/','lib64/')) else name
        expected=stack.get(canonical,item)['sha256']
        assert hashlib.sha256(Path('/' + name).read_bytes()).hexdigest() == expected, name
        if expected!=item['sha256']:derivation.append({'path':name,'build_dependency':item['sha256'],'installed_node_dependency':expected})
    record('native-runtime-derivation.json',{'unchanged_native_elf':True,'captured_public_stack':bool(stack),
                                          'dependency_overrides':derivation})
    native_boundary('c01-managed-v6', 'app-v6-missing-fd12', 'required Tor lifecycle FD12 missing')
    native_boundary('c01-managed-v7', 'app-v7-missing-fd14', 'required v7 radio FD14 missing')
    native_boundary('c01-managed-v7', 'app-v7-missing-fd12', 'required Tor lifecycle FD12 missing', slots=(14,))
    native_boundary('c01-managed-v7', 'app-v7-missing-fd11', 'required network settings FD11 missing', slots=(12,14))
    native_boundary('c01-managed-v7', 'app-v7-missing-fd10', 'required D09 FD10 missing', slots=(11,12,14))
    native_boundary('c01-managed-v6', 'app-v6-missing-fd11', 'required network settings FD11 missing', slots=(12,))
    native_boundary('c01-managed-v6', 'app-v6-missing-fd10', 'required D09 FD10 missing', slots=(11, 12))
    native_boundary('c01-managed-v6', 'app-v6-foreign-descriptors', 'D09 FD10 type/passcred', slots=(10, 11, 12))
    native_boundary('c01-managed-v6', 'app-v6-parent-is-not-d09',
                    'FD10 must name independent D09 creator', slots=(10, 11, 12), passcred=True)
    native_boundary('c01-managed-v5', 'obsolete-app-v5', 'obsolete protected application ABI')
    native_boundary('c01-managed-v4', 'obsolete-app-v4', 'obsolete protected application ABI')
    native_boundary('c01-managed-v3', 'obsolete-app-v3', 'obsolete protected application ABI')
    native_boundary('c01-provision-store-v1', 'store-provision-unconfined-root',
                    'C01 requires permanently dropped non-root credentials')

    # Execute the established prerequisite unchanged and retain its independent
    # real kernel/runtime observations. Its tmux command is NOT default terminal.
    env = dict(os.environ, C01_DISPOSABLE_SYSTEMD='1')
    p = subprocess.run(['python3', '-B', str(PACKAGE / 'tests/regression/c01-prerequisite-systemd.py')],
                       env=env, capture_output=True, text=True, timeout=300)
    (EVIDENCE / 'prerequisite.log').write_text(p.stdout + p.stderr)
    record('prerequisite-exit.json', {'exit': p.returncode, 'qualification': 'separate prerequisite only'})
    print('PREREQUISITE SUB-EXIT', p.returncode, p.stdout, p.stderr, flush=True)
    if p.returncode:
        record('result.json',{'status':'BLOCKED_PREREQUISITE','exit':77,'native_ready':False,
                             'installed_acceptance':False,'evidence':'prerequisite.log'})
        return 77

    composed = subprocess.run(['python3', '-B', str(PACKAGE / 'tests/regression/c01-compose.py')],
                              capture_output=True, text=True, timeout=600)
    (EVIDENCE / 'composition.log').write_text(composed.stdout + composed.stderr)
    print('COMPOSITION SUB-EXIT', composed.returncode, composed.stdout, composed.stderr, flush=True)
    if composed.returncode:
        record('result.json',{'status':'BLOCKED_COMPOSITION','exit':77,'native_ready':False,
                             'installed_acceptance':False,'evidence':'composition.json'})
        return 77
    if os.environ.get('C01_POWER_COMPONENT') == '1':
        assert composed.returncode == 0, 'power component prerequisites failed'
        result = subprocess.run(['python3','-B',str(PACKAGE/'tests/regression/c01-power-systemd.py')],
                                capture_output=True,text=True,timeout=180)
        (EVIDENCE/'power-test.log').write_text(result.stdout+result.stderr)
        print(result.stdout,result.stderr,flush=True)
        return result.returncode

    # Native root supervisor is itself MainPID (embedded CPython, no Python exec).
    # Type=notify ensures a successfully executed but unready process cannot pass.
    Path('/run/archipelago-c01').mkdir(exist_ok=True)
    Path('/var/lib/archipelago-c01').mkdir(mode=0o700, exist_ok=True)
    # Access, D09 and network owners authenticate this exact service identity.
    # A differently named otherwise identical unit cannot compose their leases.
    unit = Path('/etc/systemd/system/archipelago.service')
    if not unit.exists():
        unit.write_text('[Service]\nType=notify\nNotifyAccess=main\nDelegate=yes\nDelegateSubgroup=supervisor\nNoNewPrivileges=yes\n'
        'NetworkNamespacePath=/run/netns/archipelago-managed\n'
        'ExecStart=/usr/local/bin/archipelago c01-supervisor-v3\n'
        'TimeoutStartSec=150\nTimeoutStopSec=infinity\nRestart=no\n')
    assert run('systemctl', 'daemon-reload').returncode == 0
    composition = json.loads((EVIDENCE / 'composition.json').read_text())
    retained_units = ('archipelago-c01-access.service','archipelago-runtime-broker.service',
                      'archipelago-c01-network.service','archipelago-d09-lifecycle.service',
                      'archipelago-nts.service','user@1000.service')
    def owner_states():
        return {name:dict(line.split('=',1) for line in run('systemctl','show',name,
                    '-p','MainPID','-p','ActiveState','-p','InvocationID').stdout.splitlines() if '=' in line)
                for name in retained_units}
    before_owners = owner_states()
    properties=run('systemctl','show','archipelago.service','-p','CapabilityBoundingSet','-p','AmbientCapabilities',
                   '-p','SecureBits','-p','NoNewPrivileges','-p','RestrictNamespaces','-p','PrivateUsers',
                   '-p','RestrictSUIDSGID','-p','RootDirectory').stdout
    try:file_caps=os.getxattr('/usr/local/bin/archipelago','security.capability').hex()
    except OSError:file_caps=None
    record('native-execution-attributes.json',{'systemd':properties,'file_capabilities':file_caps})
    credential_samples={}
    stop_samples=threading.Event()
    def sample_credentials():
        while not stop_samples.wait(.005):
            for role in ('supervisor','c01-authority'):
                path=Path('/sys/fs/cgroup/system.slice/archipelago.service')/role/'cgroup.procs'
                try:members=path.read_text().split()
                except FileNotFoundError:continue
                for pid in members:
                    try:
                        status=dict(line.split(':',1) for line in Path('/proc',pid,'status').read_text().splitlines() if ':' in line)
                        item={k:status[k].strip() for k in ('Uid','Gid','CapInh','CapPrm','CapEff','CapBnd','CapAmb','NoNewPrivs','Seccomp')}
                        item['exe']=os.readlink('/proc/'+pid+'/exe')
                        item['apparmor']=Path('/proc',pid,'attr/apparmor/current').read_text().strip()
                        credential_samples[(role,pid,json.dumps(item,sort_keys=True))]=item
                    except (FileNotFoundError,ProcessLookupError,PermissionError):pass
    sampler=threading.Thread(target=sample_credentials,daemon=True);sampler.start()
    if composition.get('d09_owner', {}).get('status') == 'completed':
        start = run('/usr/libexec/archipelago-d09-provision', '--activate-prerequisite')
    else:
        start = run('systemctl', 'start', '--no-block', unit.name)
    observations = []
    for _ in range(1500):
        state = run('systemctl', 'show', unit.name, '-p', 'MainPID', '-p', 'ActiveState',
                    '-p', 'SubState', '-p', 'Result', '-p', 'ExecMainStatus', '-p', 'Job').stdout
        fields = dict(line.split('=', 1) for line in state.splitlines() if '=' in line)
        pid = int(fields.get('MainPID', '0'))
        if pid:
            try:
                fields['proc_exe'] = os.readlink(f'/proc/{pid}/exe')
                fields['cgroup'] = Path(f'/proc/{pid}/cgroup').read_text()
            except FileNotFoundError:
                pass
        if not observations or fields != observations[-1]:
            observations.append(fields)
        if fields.get('ActiveState') in ('failed', 'active'):
            break
        if (fields.get('ActiveState')=='inactive' and fields.get('Job') in ('','0','0 /') and
                any(item.get('MainPID','0')!='0' for item in observations)):
            break
        time.sleep(.1)
    journal = run('journalctl', '-u', unit.name, '--no-pager').stdout
    stop_samples.set();sampler.join(timeout=2)
    record('native-process-credentials.json',[{'role':role,'pid':pid,**item}
           for (role,pid,_),item in credential_samples.items()])
    (EVIDENCE / 'native-supervisor.log').write_text(journal)
    record('native-supervisor.json', {'start_exit': start.returncode, 'start_reply':start.stdout,
        'start_stderr':start.stderr, 'observations': observations})
    assert 'IO Safety violation' not in journal, 'embedding retained a descriptor across Python cleanup'
    assert observations[-1].get('Result') != 'core-dump', 'native startup failure must not abort'
    if observations[-1].get('ActiveState')=='active':
        assert all(item['status']=='completed' for item in composition.values()), 'ready despite failed prerequisites'
        assert 'C01_BASELINE_READY' in journal and 'C01_LIFETIME_CAPABILITIES_ACQUIRED' in journal
        accepted=subprocess.run(['python3','-B',str(PACKAGE/'tests/regression/c01-normal-acceptance.py')],
                                capture_output=True,text=True,timeout=850)
        (EVIDENCE/'native-acceptance.log').write_text(accepted.stdout+accepted.stderr)
        record('result.json',{'status':'PASSED_PRIVATE_BASELINE' if accepted.returncode==0 else 'BLOCKED_LIVE_ACCEPTANCE',
               'exit':accepted.returncode,'native_ready':True,'acceptance_log':'native-acceptance.log',
               'installed_acceptance':False,'optional_capabilities':json.loads((EVIDENCE/'conditional-admission.json').read_text())['optional_unavailable']})
        print('ACTUAL NATIVE ACCEPTANCE',accepted.returncode,accepted.stdout,accepted.stderr,flush=True)
        return accepted.returncode
    if all(item['status']=='completed' for item in composition.values()):
        assert 'C01_LIFETIME_CAPABILITIES_ACQUIRED' in journal, 'complete provisioning did not reach native lifetime acquisition'
        if (PACKAGE / 'candidate-appointment.json').exists():
            record('candidate-startup-outcome.json', {'active':False,
                   'timing_bound_refusal':'invalid platform qualification bound' in journal,
                   'native_journal':'native-supervisor.log', 'composed_qualification':False})
        else:
            assert 'invalid platform qualification bound' in journal, 'unexpected earlier native composition failure'
    if composition.get('d09_owner',{}).get('status')=='completed':
        ledger_path=Path('/var/lib/archipelago-d09/ledger.json')
        deadline=time.monotonic()+10
        while time.monotonic()<deadline:
            ledger=json.loads(ledger_path.read_text())
            if ledger.get('prerequisite',{}).get('state')=='START_UNCERTAIN': break
            time.sleep(.1)
        record('d09-owner-final.json',ledger)
        assert ledger['prerequisite']['state']=='START_UNCERTAIN', 'D09 must fence failed normal startup'
        assert len(ledger['grants'])==1 and ledger['receipt'] is None, 'failed prerequisite repeated or became APPLY'
    access_journal=Path('/var/lib/archipelago-c01/access/journal.json')
    if access_journal.exists():
        record('access-owner-final.json',json.loads(access_journal.read_text()))
    run('systemctl', 'stop', unit.name)
    time.sleep(6)  # cross the access owner's five-second checker period
    after_owners = owner_states()
    record('retained-owner-lifetimes.json', {'before':before_owners,'after':after_owners,
        'scope':'actual owner lifetimes across failed native activation; not APP_READY'})
    work=Path('/var/lib/c01-fixture/home/Work/operator.txt')
    meta=work.stat()
    preserved={'device':meta.st_dev,'inode':meta.st_ino,'uid':meta.st_uid,'gid':meta.st_gid,
               'mode':meta.st_mode & 0o7777,'sha256':hashlib.sha256(work.read_bytes()).hexdigest()}
    record('work-after.json',preserved)
    assert preserved==json.loads((EVIDENCE/'work-before.json').read_text()), 'original Work changed during composed transition/start failure'
    if all(composition.get(name,{}).get('status')=='completed' for name in
           ('access_owner','runtime_owner','network_owner','d09_owner','nts_owner')):
        assert before_owners == after_owners and all(v['ActiveState']=='active' and int(v['MainPID'])>1
            for v in after_owners.values()), 'independent owner/runtime lifetime lost on management failure'
    # Record the exact frozen implementation of the outer gate, never replace it.
    text = (PACKAGE / 'scripts/c01_supervisor.py').read_text()
    gate = text[text.index('def admit('):text.index('def acquire_admission_leases(')]
    (EVIDENCE / 'outer-admission-source.txt').write_text(gate)
    record('result.json', {
        'status': 'BLOCKED', 'exit': 77, 'prerequisite_exit': p.returncode,
        'native_default_terminal': 'not reached',
        'actual_owner_leases': 'acquired and retired by actual native supervisor' if
            'C01_LIFETIME_CAPABILITIES_ACQUIRED' in journal else 'not established by this fixture',
        'clock_storage_bootstrap': {'store_provisioning':composition.get('stores',{}).get('status'),
                                    'clock_recovery':'not reached'},
        'generation_A_B_C_B': 'not reached',
        'recovery_restart': 'not reached', 'physical_platform_qualification': 'not performed',
        'native_failure_lines': [line for line in journal.splitlines()
                                 if 'Error:' in line or 'C01_ADMISSION_UNAVAILABLE' in line],
        'composition_stage_failures': {name:item.get('error') for name,item in composition.items()
                                       if item['status']!='completed'},
        'blockers': ['See native-supervisor.log for the actual earliest startup boundary'],
        'conditional_candidate_bound':(EVIDENCE/'candidate-binding.json').exists(),
        'conditional_admission_measured':'C01_CONDITIONAL_ADMISSION_HELD' in journal,
        'source_boundary': 'ordinary native binary and selectors; prerequisite tmux is not native terminal readiness'})
    print('BLOCKED normal native composition: see result.json and native-supervisor.log', flush=True)
    return 77


if __name__ == '__main__':
    raise SystemExit(main())