import datetime, hashlib, json, os, re, signal, subprocess, sys, time
from pathlib import Path
ROOT = Path('/run/media/kmosoti/data/FabricO11y')
REPO = ROOT / 'scratch/console-access-acceptance'
OUT = ROOT / 'results/service-opening-candidate'
STORE = ROOT / 'scratch/service-opening-census'
FROZEN = 'a74ab66bae81197d8f7a22f61ccba1a686b7a79e'
def require(value, reason):
    if not value:
        raise ValueError(reason)
label, reserve_text, *command = sys.argv[1:]
reserve = int(reserve_text)
delegated = command[:1] == ['--delegate']
if delegated: command = command[1:]
require(re.fullmatch(r'[a-z0-9-]{1,80}', label) and reserve > 0 and command, 'ordinary workload admission guard')
record_path = OUT / (label + '.json')
require(not record_path.exists(), 'unique run label required')
def census():
    # Rootless container files can belong to subordinate UIDs. Inspect all of
    # them in the same user namespace; never skip unreadable directories.
    require(not STORE.is_symlink(), 'ordinary workload admission guard')
    require((STORE / '.owned-census-store').read_text() == 'service-opening-census-v1\n', 'ordinary workload admission guard')
    args = ['podman', '--root', str(STORE/'graph'), '--runroot', str(STORE/'run'),
            '--tmpdir', str(STORE/'tmp'), '--storage-driver', 'vfs',
            'unshare', 'python3', '-B', str(OUT/'census.py')]
    env = dict(os.environ, TMPDIR=str(STORE/'tmp'), TMP=str(STORE/'tmp'), TEMP=str(STORE/'tmp'))
    result = subprocess.run(args, env=env, text=True, capture_output=True, timeout=25)
    if result.returncode:
        raise RuntimeError(f'strict namespace census failed ({result.returncode}): {result.stderr}')
    observed = json.loads(result.stdout)
    require(observed['uid'] == 0 and observed['gid'] == 0, 'ordinary workload admission guard')
    require(type(observed['data_bytes']) is int and observed['data_bytes'] >= 0, 'ordinary workload admission guard')
    return observed
def git(*args):
    return subprocess.check_output(['git', *args], cwd=REPO, text=True).strip()
require(git('rev-parse', 'HEAD') == FROZEN and not git('status', '--porcelain'), 'clean frozen source required')
require(not subprocess.check_output(['systemctl', '--user', 'list-units', 'fabric-work-*', '--state=running', '--plain', '--no-legend'], text=True).strip(), 'another project workload is active')
initial = census()
used = initial['data_bytes']
record = {'label': label, 'command': command, 'delegated': delegated, 'frozen': None, 'scope': 'harness controls only; no candidate acceptance evidence', 'harness_revision': FROZEN, 'source_status_before': git('status', '--porcelain'), 'storage_root': str(ROOT), 'admission_bytes': used, 'admission_observation': initial, 'reserved_bytes': reserve, 'observations': [], 'started_utc': datetime.datetime.now(datetime.timezone.utc).isoformat()}
if used + reserve > 95_000_000_000:
    record.update(state='not_run', reason='95 GB storage admission threshold')
    record_path.write_text(json.dumps(record, indent=2)+'\n')
    raise SystemExit('storage admission denied; project command not launched')
receipts = REPO / 'target/resource-containment/runs'
before = set(receipts.glob('*.json'))
env = dict(os.environ, FABRIC_STORAGE_ROOT=str(ROOT), CARGO_HOME=str(ROOT/'toolchain-cache/cargo'), RUSTUP_HOME=str(ROOT/'toolchain-cache/rustup'), CARGO_BUILD_JOBS='2')
env['PATH'] = str(ROOT/'tools/candidate-dpkg/bin') + ':' + env['PATH']
log_path = OUT / (label + '.log')
record.update(state='running', log=str(log_path))
record_path.write_text(json.dumps(record, indent=2)+'\n')
def find_unit():
    with log_path.open() as stream:
        prefix = stream.read(65536)
    match = re.search(r'Running as unit: (fabric-work-[0-9a-f]+)\.service', prefix)
    return match.group(1) if match else None
def stop_owned(process):
    unit = find_unit()
    if unit:
        stopped = subprocess.run(['systemctl', '--user', 'stop', unit],
                                 text=True, capture_output=True, timeout=25)
        record['stop'] = {'unit': unit, 'exit': stopped.returncode, 'stderr': stopped.stderr}
        if stopped.returncode:
            # Interrupt the launcher too; its finally block independently
            # confirms that this exact service has no surviving descendants.
            process.send_signal(signal.SIGINT)
    else:
        process.send_signal(signal.SIGINT)
        record['stop'] = {'unit': None, 'launcher_interrupted': True}
with log_path.open('wb') as log:
    launcher = ['python3', '-B', 'tools/resource_group.py', *(['--delegate'] if delegated else []), '--', *command]
    process = subprocess.Popen(launcher, cwd=REPO, env=env, stdout=log, stderr=subprocess.STDOUT)
    next_census = time.monotonic() + 30
    try:
        while process.poll() is None:
            time.sleep(1)
            if time.monotonic() >= next_census:
                observed = census()
                record['observations'].append({'unix':time.time(), **observed})
                record_path.write_text(json.dumps(record, indent=2)+'\n')
                if observed['data_bytes'] >= 95_000_000_000:
                    record['interruption'] = '95 GB threshold reached'
                    break
                next_census = time.monotonic() + 30
    except BaseException as error:
        record['interruption'] = f'{type(error).__name__}: {error}'
    finally:
        if record.get('interruption') and process.poll() is None:
            try:
                stop_owned(process)
            except BaseException as error:
                record['stop_error'] = f'{type(error).__name__}: {error}'
                process.send_signal(signal.SIGINT)
        status = process.wait()
record['launcher_exit'] = status
if record.get('interruption'): status = 2
with log_path.open('rb') as stream:
    log_digest = hashlib.file_digest(stream, 'sha256').hexdigest()
record.update(state='interrupted' if record.get('interruption') else 'completed', exit=status, source_status_after=git('status', '--porcelain'), finished_utc=datetime.datetime.now(datetime.timezone.utc).isoformat(), log_sha256=log_digest, resource_receipts=[])
archive = OUT/'resource-containment'; archive.mkdir(exist_ok=True)
for p in sorted(set(receipts.glob('*.json')) - before):
    value=json.loads(p.read_text())
    if value['command'] != command: continue
    q=archive/p.name; q.write_bytes(p.read_bytes())
    record['resource_receipts'].append({'path':str(q),'sha256':hashlib.sha256(q.read_bytes()).hexdigest(),'exit':value['exit'],'stop_confirmed':value['stop_confirmed'],'retained_failure_evidence':value['retained_failure_evidence']})
if not record['resource_receipts'] or any(not r['stop_confirmed'] or r['exit'] != status for r in record['resource_receipts']):
    record['receipt_error'] = 'missing or inconsistent outer resource receipt'
    status = 2
    record['exit'] = status
record_path.write_text(json.dumps(record, indent=2)+'\n')
print(json.dumps(record, indent=2))
if status: print(log_path.read_text()[-8000:])
raise SystemExit(status)

"""Disposable target-only clock fixture and contained native binding admission."""
import argparse
import ctypes
import hashlib
import json
import os
from pathlib import Path
import platform
import select
import shutil
import signal
import socket
import subprocess
import sys
import tempfile
import threading
import time
import session_clock_oracle as oracle

HERE=Path(__file__).resolve().parent
sys.path.insert(0,str(HERE.parent))
from resource_group import STORAGE, require_limits


def digest(path):
    with Path(path).open('rb') as stream:return hashlib.file_digest(stream,'sha256').hexdigest()


class KernelTime(ctypes.Structure):_fields_=[('sec',ctypes.c_long),('nsec',ctypes.c_long)]
def kernel_time():
    value=KernelTime();oracle.require(ctypes.CDLL(None,use_errno=True).syscall(228,0,ctypes.byref(value))==0,'fixture-raw-clock');return value.sec*1000000000+value.nsec


def freeze_images(executable,helper):
    frozen={}
    for name,path in (('executable',executable),('helper',helper)):
        before=path.stat();frozen[name+'_sha256']=digest(path);after=path.stat()
        oracle.require((before.st_dev,before.st_ino,before.st_size,before.st_mtime_ns,before.st_ctime_ns)==(after.st_dev,after.st_ino,after.st_size,after.st_mtime_ns,after.st_ctime_ns),'isolation-'+('target' if name=='executable' else 'helper')+'-frozen')
        frozen[name+'_identity']={'device':before.st_dev,'inode':before.st_ino}
    return frozen


def process_binding(pid, executable, helper, intercepted=True, *, expected=None):
    proc=Path(f'/proc/{pid}');actual=proc/'exe'
    if expected is not None:
        for name,paths,message in (('executable',(actual,executable),'isolation-target-frozen'),('helper',(helper,),'isolation-helper-frozen')):
            for path in paths:
                stat=path.stat()
                oracle.require({'device':stat.st_dev,'inode':stat.st_ino}==expected[name+'_identity'] and digest(path)==expected[name+'_sha256'],message)
    oracle.require(actual.resolve()==executable.resolve() and digest(actual)==digest(executable),'isolation-target')
    rows=[line.split(maxsplit=5) for line in (proc/'maps').read_text().splitlines()]
    mappings={(row[3],row[4],row[5]) for row in rows if len(row)==6 and row[5]==str(helper)};stat=helper.stat()
    if expected is not None and mappings:
        frozen=expected['helper_identity']
        oracle.require(mappings=={(f'{os.major(frozen["device"]):02x}:{os.minor(frozen["device"]):02x}',str(frozen['inode']),str(helper))},'isolation-helper-frozen')
    oracle.require(mappings==({(f'{os.major(stat.st_dev):02x}:{os.minor(stat.st_dev):02x}',str(stat.st_ino),str(helper))} if intercepted else set()),'isolation-interception')
    libc=next(Path(row[5]) for row in rows if len(row)==6 and row[5].endswith('/libc.so.6'))
    return {'pid':pid,'executable_sha256':digest(actual),'helper_sha256':digest(helper) if mappings else None,'libc':str(libc),'libc_sha256':digest(libc),
            'start_ticks':int((proc/'stat').read_text().rsplit(')',1)[1].split()[19]),'cgroup':(proc/'cgroup').read_text(),'helper_maps':sorted(mappings)}


def stop_group(process):
    try:os.killpg(process.pid,signal.SIGTERM)
    except ProcessLookupError:process.wait(timeout=3);return
    try:process.wait(timeout=3)
    except subprocess.TimeoutExpired:pass
    try:os.killpg(process.pid,signal.SIGKILL)
    except ProcessLookupError:pass
    process.wait(timeout=3);deadline=time.monotonic()+3
    while True:
        try:
            while os.waitpid(-process.pid,os.WNOHANG)[0]:pass
        except ChildProcessError:pass
        try:os.killpg(process.pid,0)
        except ProcessLookupError:return
        oracle.require(time.monotonic()<deadline,'fixture-process-group-stop');time.sleep(.01)


class SessionClock:
    def __init__(self, executable, helper, directory, deadline, initial=None, socket_type=socket.SOCK_SEQPACKET, preload=True, invalid_fd=False):
        self.deadline=deadline;self.frames=[];self.rows=[];self.bytes=0;self.error=None;self.buffer=b''
        env=dict(os.environ);oracle.require(not any(name in env for name in ('LD_PRELOAD','LD_AUDIT','FABRIC_CLOCK_FD')),'fixture-environment')
        self.parent,self.child=socket.socketpair(socket.AF_UNIX,socket_type|socket.SOCK_NONBLOCK|socket.SOCK_CLOEXEC)
        self.fd=self.child.fileno();self.socket_identity=os.readlink(f'/proc/self/fd/{self.fd}')
        self.parent.send(initial if initial is not None else self.frame(0));self.stderr=directory.open('xb')
        if preload:env.update(LD_PRELOAD=str(helper),FABRIC_CLOCK_FD=str(1048575 if invalid_fd else self.fd))
        self.argv=[str(executable),'interactive',str(self.fd),str(helper),self.socket_identity]
        try:self.process=subprocess.Popen(self.argv,env=env,pass_fds=(self.fd,),stdin=subprocess.PIPE,stdout=subprocess.PIPE,stderr=self.stderr,start_new_session=True)
        except BaseException:self.parent.close();self.child.close();self.stderr.close();raise
        self.child.close();self.receiver=threading.Thread(target=self.receive,daemon=True);self.receiver.start()
    def frame(self,offset):
        frame=b'FCLKv1\0\0'+len(self.frames).to_bytes(8,'little')+offset.to_bytes(8,'little')
        oracle.check_frames(self.frames+[frame]);self.frames.append(frame);return frame
    def advance(self,offset):oracle.require(self.parent.send(self.frame(offset))==24,'fixture-control-send')
    def receive(self):
        try:
            while True:
                if not select.select([self.parent],[],[],.1)[0]:continue
                raw=self.parent.recv(257)
                if not raw:return
                self.bytes+=len(raw);oracle.require(len(raw)<=256 and self.bytes<=1048576,'fixture-ledger-bound')
                self.rows.append(json.loads(raw))
        except BaseException as error:
            self.error=type(error).__name__+': '+str(error)
            try:os.killpg(self.process.pid,signal.SIGTERM)
            except ProcessLookupError:pass
    def line(self):
        while b'\n' not in self.buffer:
            oracle.require(self.error is None,'fixture-ledger-error')
            remaining=self.deadline-time.monotonic();oracle.require(remaining>0,'fixture-deadline')
            oracle.require(select.select([self.process.stdout],[],[],remaining)[0],'fixture-deadline')
            raw=os.read(self.process.stdout.fileno(),4096);oracle.require(bool(raw),'fixture-probe-eof')
            self.buffer+=raw;oracle.require(len(self.buffer)<=65536,'fixture-probe-bound')
        raw,self.buffer=self.buffer.split(b'\n',1);return json.loads(raw)
    def command(self,name):self.process.stdin.write((name+'\n').encode());self.process.stdin.flush()
    def close(self):
        try:stop_group(self.process)
        finally:
            self.receiver.join(timeout=1);self.parent.close();self.process.stdin.close();self.process.stdout.close();self.stderr.close()
        oracle.require(not self.receiver.is_alive(),'fixture-receiver-stop')
        try:os.killpg(self.process.pid,0)
        except ProcessLookupError:return
        raise ValueError('fixture-process-group-stop')


def native_admission(executable,helper,output,deadline,report):
    def close(current,evidence):
        try:current.close()
        except BaseException as error:
            report['native_cleanup_error']=type(error).__name__+': '+str(error);raise
        finally:
            evidence['ledger']=list(current.rows);evidence['frames']=[frame.hex() for frame in current.frames]
    target=SessionClock(executable,helper,output/'positive.stderr',deadline);report['clocks']=[];report['system_time']=[]
    try:
        ready=target.line();report['ready']=ready
        oracle.require(ready=={'kind':'ready','pid':target.process.pid,'env_clear':True,'cloexec':True,'socket_type':5} and os.readlink(f'/proc/{target.process.pid}/fd/{target.fd}')==target.socket_identity and int(Path(f'/proc/{target.process.pid}/fdinfo/{target.fd}').read_text().split('flags:',1)[1].split()[0],8)&os.O_CLOEXEC,'isolation-control')
        report['target']=process_binding(target.process.pid,executable,helper,expected=report['frozen_images']);report['controller']=process_binding(os.getpid(),Path(sys.executable),helper,False);report['host_clocks']=[]
        report['control_fd']={'identity':target.socket_identity,'fdinfo':Path(f'/proc/{target.process.pid}/fdinfo/{target.fd}').read_text()};report['probe_argv']=target.argv
        for sequence,offset in enumerate((0,1,1799,1800,28799,28800)):
            if sequence:target.advance(offset)
            if sequence==2:
                target.command('fork');fork,child,done=(target.line() for _ in range(3));report['fork']=[fork,child,done]
                row=fork['row'];oracle.require(row['pid']==child['pid']==done['fork_pid']!=target.process.pid and done['status']==0,'isolation-fork-pid')
                oracle.require(row['offset_s']==1 and row['sequence']==done['sequence_before']==1 and row['returned_ns']==row['raw_ns'],'isolation-fork-consumption')
                oracle.require(row['syscall_before_ns']<=row['raw_ns']<=row['syscall_after_ns'] and row['syscall_after_ns']-row['syscall_before_ns']<=1000000000,'isolation-host-clock')
                oracle.require(child['kind']=='exec' and child['env_clear'] and not child['helper_mapped'] and not child['control_fd_present'],'isolation-exec-escape')
                oracle.require(child['syscall_before_ns']<=child['clock_ns']<=child['syscall_after_ns'] and child['syscall_after_ns']-child['syscall_before_ns']<=1000000000,'isolation-host-clock')
            target.command('clock')
            for _ in range(3):
                value=target.line();oracle.require(value['kind']=='clock','clock-binding');report['clocks'].append(value['row'])
            system=target.line();oracle.require(system['kind']=='system','clock-binding');report['system_time'].append(system['row'])
            before=kernel_time();clock=time.time_ns();after=kernel_time();oracle.require(before<=clock<=after and after-before<=1000000000 and (not report['host_clocks'] or before>=report['host_clocks'][-1]['syscall_after_ns']),'isolation-host-clock');report['host_clocks'].append({'sequence':sequence,'syscall_before_ns':before,'clock_ns':clock,'syscall_after_ns':after})
        target.command('sleep');sleep=target.line();oracle.require(sleep['kind']=='sleep','sleep-fields');report['sleep']=sleep['row']
        oracle.check_frames(target.frames);oracle.check_probe_clocks(report['clocks'],target.process.pid);oracle.check_sleep(report['sleep'])
        for row,clock in zip(report['system_time'],report['clocks'][::3]):
            oracle.require((row['pid'],row['sequence'],row['offset_s'],row['clock_id'])==(clock['pid'],clock['sequence'],clock['offset_s'],0),'clock-binding')
            oracle.require(row['syscall_before_ns']<=row['raw_ns']<=row['syscall_after_ns'] and row['syscall_after_ns']-row['syscall_before_ns']<=1000000000,'clock-raw-bracket')
            oracle.require(row['returned_ns']==row['raw_ns']+row['offset_s']*1000000000,'clock-realtime-only')
        target.command('quit');oracle.require(target.process.wait(timeout=max(.001,deadline-time.monotonic()))==0,'fixture-probe-exit')
    finally:
        close(target,report)
    oracle.require(target.error is None,'fixture-ledger-error')
    for row in report['clocks']+report['system_time']:
        oracle.require({key:value for key,value in row.items() if not key.startswith('syscall_')} in report['ledger'],'clock-ledger-binding')
    report['controls']=[]
    initial=b'FCLKv1\0\0'+bytes(16)
    controls=[('short',initial[:-1],'FCLK-FRAME'),('oversize',initial+b'x','FCLK-FRAME'),('version',b'bad-v001'+bytes(16),'FCLK-FRAME'),
              ('initial-sequence',initial[:8]+(1).to_bytes(8,'little')+bytes(8),'FCLK-SEQUENCE'),('initial-offset',initial[:16]+(1).to_bytes(8,'little'),'FCLK-OFFSET'),
              ('skip',None,'FCLK-SEQUENCE'),('duplicate',None,'FCLK-SEQUENCE'),('decrease',None,'FCLK-OFFSET'),('cap',None,'FCLK-OFFSET'),('overflow',None,'FCLK-OFFSET'),
              ('count',None,'FCLK-COUNT'),('death',None,'FCLK-EOF'),('closed',None,'FCLK-RECV'),('descriptor',None,'FCLK-FD'),('type',None,'FCLK-TYPE'),('unpreloaded',None,'isolation-interception')]
    for name,packet,expected in controls:
        error_path=output/(name+'.stderr');current=SessionClock(executable,helper,error_path,deadline,initial=packet,socket_type=socket.SOCK_STREAM if name=='type' else socket.SOCK_SEQPACKET,preload=name!='unpreloaded',invalid_fd=name=='descriptor')
        evidence={'case':name,'refusal':expected,'exit':None,'stderr':None};report['controls'].append(evidence)
        try:
            if name=='unpreloaded':
                oracle.require(current.line()=={'kind':'absent','pid':current.process.pid},'fixture-unpreloaded')
                try:process_binding(current.process.pid,executable,helper,expected=report['frozen_images'])
                except ValueError as error:oracle.require(str(error)==expected,'fixture-wrong-refusal')
                else:raise ValueError('fixture-unpreloaded-admitted')
            else:
                if packet is None and name not in ('type','descriptor'):
                    current.line()
                    if name=='death':current.parent.shutdown(socket.SHUT_WR)
                    elif name=='decrease':current.parent.send(initial[:8]+(1).to_bytes(8,'little')+(1).to_bytes(8,'little'));current.parent.send(initial[:8]+(2).to_bytes(8,'little')+bytes(8))
                    else:
                        values=[(i,0) for i in range(1,65)] if name=='count' else [(2 if name=='skip' else 0 if name=='duplicate' else 1,28861 if name=='cap' else 2**64-1 if name=='overflow' else 0)]
                        if name!='closed':
                            for seq,value in values:current.parent.send(initial[:8]+seq.to_bytes(8,'little')+value.to_bytes(8,'little'))
                    current.command('close' if name=='closed' else 'clock')
                oracle.require(current.process.wait(timeout=max(.001,deadline-time.monotonic()))==78,'fixture-control-exit')
        finally:close(current,evidence)
        stderr=error_path.read_text();oracle.require(name=='unpreloaded' or stderr.strip()==expected,'fixture-wrong-refusal')
        evidence.update(exit=current.process.returncode,stderr=stderr)


def main():
    parser=argparse.ArgumentParser(description=__doc__);parser.add_argument('--output',type=Path,required=True);parser.add_argument('--rustc',required=True);parser.add_argument('--cc',default='gcc');args=parser.parse_args()
    require_limits();oracle.require(platform.system()=='Linux' and platform.machine()=='x86_64','fixture-platform')
    output=args.output;oracle.require(output.is_absolute() and output.resolve()==output and output.is_relative_to(STORAGE),'fixture-output');output.mkdir(mode=0o700)
    scratch=Path(os.environ['FABRIC_SCRATCH_ROOT']);oracle.require(scratch.is_dir() and scratch.is_relative_to(STORAGE),'fixture-scratch')
    directory=Path(tempfile.mkdtemp(prefix='session-clock-',dir=scratch));deadline=time.monotonic()+240;report={'scope':'native binding only; actual packaged browser session checks pending','state':'failed','exit':1,'commands':[]}
    def run(command):
        report['commands'].append(command);path=output/('tool-'+str(len(report['commands']))+'.log');timeout=max(.001,deadline-time.monotonic());failure=None
        with path.open('xb') as stream:
            process=subprocess.Popen(command,stdout=stream,stderr=subprocess.STDOUT,start_new_session=True)
            try:
                while process.poll() is None:
                    if time.monotonic()>=deadline:raise subprocess.TimeoutExpired(command,timeout)
                    oracle.require(path.stat().st_size<=1048576,'fixture-tool-output');time.sleep(.01)
            except BaseException as error:failure=error;raise
            finally:
                try:stop_group(process)
                except BaseException as error:
                    report['tool_cleanup_error']=type(error).__name__+': '+str(error)
                    if failure is None:raise
                finally:
                    with path.open('rb') as captured:raw=captured.read(1048576)
                    report.setdefault('tool_outputs',[]).append({'command':command,'exit':process.returncode,'raw_log':str(path),'output':raw.decode(errors='replace')})
                    if isinstance(failure,subprocess.TimeoutExpired):failure.output=raw
        oracle.require(path.stat().st_size<=1048576,'fixture-tool-output');oracle.require(process.returncode==0,'fixture-tool-exit');return raw.decode()
    try:
        compiler=Path(shutil.which(args.cc)).resolve();rustc=Path(shutil.which(args.rustc)).resolve();version=run([str(rustc),'-vV'])
        oracle.require('release: 1.99.0' in version and 'commit-hash: b940084d7' in version,'fixture-toolchain')
        run([str(compiler),'--version']);helper=directory/'session-clock.so';executable=directory/'session-clock-probe'
        run([str(compiler),'-shared','-fPIC','-pthread','-O2','-Wall','-Wextra','-Werror','-o',str(helper),str(HERE/'session_clock.c'),'-ldl'])
        run([str(rustc),'--edition=2021','-C','opt-level=1','-o',str(executable),str(HERE/'session_clock_probe.rs')])
        dynamic=run(['readelf','--dyn-syms','--wide',str(executable)]);oracle.require('clock_gettime@' in dynamic,'fixture-dynamic-clock-import')
        interpreter=run(['readelf','-lW',str(executable)]).split('Requesting program interpreter: ',1)[1].split(']',1)[0]
        report['frozen_images']=freeze_images(executable,helper)
        report['hashes']={str(path):digest(path) for path in (compiler,rustc,helper,executable,Path(interpreter),HERE/'session_clock.py',HERE/'session_clock.c',HERE/'session_clock_probe.rs')}
        oracle.require(all(report['hashes'][str(path)]==report['frozen_images'][name+'_sha256'] for name,path in (('executable',executable),('helper',helper))),'fixture-image-hash')
        oracle.require(sum(path.stat().st_size for path in directory.rglob('*') if path.is_file())<=512*1024**2,'fixture-storage-bound')
        native_admission(executable,helper,output,deadline,report)
        report['build_id_notes']={str(path):run(['readelf','-nW',str(path)]) for path in (helper,executable,Path(interpreter),Path(report['target']['libc']))};report.update(state='completed',exit=0)
    except BaseException as error:report['error']=type(error).__name__+': '+str(error);raise
    finally:
        try:
            if report['state']=='completed':shutil.rmtree(directory)
            elif any(name in report for name in ('tool_cleanup_error','native_cleanup_error')):report['failure_evidence']=str(directory)
            else:
                retained=output/'failed-fixture';shutil.move(str(directory),str(retained));report['failure_evidence']=str(retained)
        except BaseException as error:
            report['cleanup_error']=type(error).__name__+': '+str(error);report['failure_evidence']=str(directory)
            if 'error' not in report:report.update(state='failed',exit=1);raise
        finally:
            report['cleanup']={'owned_fixture':str(directory),'absent':not directory.exists(),'failure_evidence':report.get('failure_evidence')}
            (output/'receipt.json').write_text(json.dumps(report,indent=2)+'\n');print(str(output/'receipt.json'))
    return report['exit']


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

"""Named rejection must survive provenance and environment checks."""
import ast
import importlib.util
import os
from pathlib import Path
import unittest
import subprocess
import tempfile
from types import SimpleNamespace
from unittest.mock import call, patch

SPEC = importlib.util.spec_from_file_location('install_qemu', Path(__file__).with_name('run-qemu.py'))
MODULE = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(MODULE)


class MutationGradingTests(unittest.TestCase):
    def test_expected_rpm_mutation_cleans_only_confirmed_stopped_scratch(self):
        main = next(node for node in ast.parse(Path(MODULE.__file__).read_text()).body
                    if isinstance(node, ast.FunctionDef) and node.name == 'main')
        final = next(node.finalbody for node in main.body if isinstance(node, ast.Try))
        index = next(i for i, node in enumerate(final) if isinstance(node, ast.Assign)
                     and any(isinstance(t, ast.Name) and t.id == 'completed_mutation' for t in node.targets))
        code = compile(ast.Module(body=final[index:index + 2], type_ignores=[]), '<actual-cleanup>', 'exec')
        for output, stopped, error, expected in (
                ('ACCEPT A6 FAIL injected', True, None, True),
                ('ACCEPT A12 FAIL unrelated', True, None, False),
                ('ACCEPT A6 FAIL injected', False, None, False),
                ('ACCEPT A6 FAIL injected', True, 'helper hash mismatch', False)):
            with self.subTest(output=output, stopped=stopped, error=error), \
                    tempfile.TemporaryDirectory(dir=os.environ['FABRIC_SCRATCH_ROOT']) as root:
                namespace = dict(args=SimpleNamespace(mutate=None, expected_mutation='root-user'),
                    grading_mutation='root-user', mutation_package_archive=None,
                    mutation_rejected=MODULE.mutation_rejected, rc=1, acceptance_log=output,
                    error=error, cleanup_ok=stopped, process_cleanup_confirmed=stopped,
                    work_dir=Path(root), result_dir=Path(root), shutil=MODULE.shutil, json=MODULE.json,
                    guest_acceptance_exit=1, mutation_package_sha256=None)
                with patch.object(MODULE.shutil, 'rmtree') as remove:
                    exec(code, namespace)
                self.assertEqual(remove.call_args_list, [call(Path(root))] if expected else [])

    def test_reboot_witness_runs_after_privilege_transition(self):
        # Origin: Fedora upgrade03 really rebooted, but SSH flattened bash -c
        # argv, expanding protected paths too early and leaving a bare test.
        def fake_ssh(args, *, input_text=None, **kwargs):
            shell = '''sudo() { privileged=yes; "$@"; }
cat() {
  [ "$privileged" = yes ] || return 1
  case "$1" in
    /proc/sys/kernel/random/boot_id) echo current ;;
    /root/accept/lifecycle-boot-id) echo "$saved" ;;
    *) return 1 ;;
  esac
}
export -f cat
export privileged saved
'''
            return subprocess.run(['bash', '-c', shell + ' '.join(args[1:])],
                                  input=input_text, text=True, capture_output=True,
                                  env={'PATH': '/usr/bin:/bin', 'saved': saved})
        with patch.object(MODULE, 'run', fake_ssh):
            saved = 'before'
            self.assertEqual(MODULE.changed_boot(['fake-ssh']).returncode, 0)
            saved = 'current'
            self.assertNotEqual(MODULE.changed_boot(['fake-ssh']).returncode, 0)
            saved = 'before'
            old = fake_ssh(['fake-ssh', 'sudo', 'bash', '-c',
                           'test "$(cat /proc/sys/kernel/random/boot_id)" != "$(cat /root/accept/lifecycle-boot-id)"'])
            self.assertNotEqual(old.returncode, 0)

    def test_all_registered_named_defects_are_required(self):
        for mutation, checks in MODULE.MUTATION_EXPECTED_FAILURES.items():
            output = '\n'.join(f'ACCEPT {check} FAIL injected' for check in sorted(checks))
            self.assertTrue(MODULE.mutation_rejected(mutation, 1, output))
            for check in checks:
                with self.subTest(mutation=mutation, missing=check):
                    self.assertFalse(MODULE.mutation_rejected(
                        mutation, 1, output.replace(f'ACCEPT {check} FAIL', f'ACCEPT {check} PASS')))

    def test_unrelated_failure_is_not_detection(self):
        self.assertFalse(MODULE.mutation_rejected('root-user', 1, 'ACCEPT A12 FAIL timeout'))

    def test_bad_provenance_or_environment_cannot_pass(self):
        output = 'ACCEPT A6 FAIL injected'
        self.assertFalse(MODULE.mutation_rejected('root-user', 1, output, 'helper hash mismatch'))
        self.assertFalse(MODULE.mutation_rejected('root-user', 1, output + '\nACCEPT A8h NOT-RUN unavailable'))
        for code in (None, 0, 2, 3, -9):
            self.assertFalse(MODULE.mutation_rejected('root-user', code, output))


class InstallationShellControls(unittest.TestCase):
    """Origin: C-035 / C-aud-7; real shell decisions, isolated paths and command stubs."""
    def shell(self, program, family='fedora', *, unit='', state='inactive', avc='',
              retained=True, purge_keeps_state=False, systemctl_mode='normal', grep_error=False):
        with tempfile.TemporaryDirectory(dir=os.environ['FABRIC_SCRATCH_ROOT']) as directory:
            root = Path(directory)
            paths = {old: root / name for old, name in (
                ('/root/accept', 'accept'), ('/var/lib/fabrico11y', 'state'),
                ('/etc/fabrico11y', 'config'), ('/usr/bin/fabric-server', 'absent-binary'))}
            work, data, config = (paths[p] for p in ('/root/accept', '/var/lib/fabrico11y', '/etc/fabrico11y'))
            work.mkdir(); config.mkdir()
            if retained: (data / 'server').mkdir(parents=True)
            account = 'fabricolly:x:995:995:Fabric O11y service::/usr/sbin/nologin\n'
            (work / 'lifecycle-account').write_text(account)
            (config / 'server.conf').write_text('synthetic configuration\n')
            for old, new in paths.items(): program = program.replace(old, str(new))
            stubs = '''
rpm() { [ "$*" = '-e fabrico11y' ]; }
dpkg() {
  if [ "$1" = -P ]; then
    if [ "$PURGE_KEEP_STATE" != yes ]; then rm -rf "$DATA"; fi
    rm -rf "$CONFIG"
  fi
}
rm() {
  [ "$1" = -rf ] || return 97; shift
  for path; do
    case "$path" in "$DATA"|"$CONFIG") command rm -rf -- "$path" ;; *) return 97 ;; esac
  done
}
package_remove() { return 0; }
getent() { printf '%s' "$ACCOUNT"; }
getenforce() { echo Enforcing; }
auditctl() { echo 'enabled 1'; }
ausearch() { printf '%s\\n' "$AVC"; }
grep() {
  if [ "$GREP_ERROR" = yes ] && [ "$1" = -E ]; then
    echo 'synthetic audit read error' >&2; return 2
  fi
  command grep "$@"
}
systemctl() {
  case "$SYSTEMCTL_MODE" in
    empty) return 3 ;;
    incomplete) echo inactive; return 3 ;;
    error) printf 'inactive\\ninactive\\n'; echo 'synthetic systemctl query error' >&2; return 1 ;;
  esac
  local rc=3 value unit
  for unit; do
    case "$unit" in fabrico11y-*.service)
      value=inactive; if [ "$unit" = "$BAD_UNIT" ]; then value=$BAD_STATE; fi
      case " $* " in *' --quiet '*) ;; *) printf '%s\\n' "$value" ;; esac
      if [ "$value" = active ]; then rc=0; fi
    esac
  done
  return "$rc"
}
'''
            env = dict(os.environ, ACCOUNT=account, DATA=str(data), CONFIG=str(config),
                       BAD_UNIT=unit, BAD_STATE=state, AVC=avc,
                       PURGE_KEEP_STATE='yes' if purge_keeps_state else 'no',
                       SYSTEMCTL_MODE=systemctl_mode, GREP_ERROR='yes' if grep_error else 'no')
            return subprocess.run(['bash', '-s', '--', 'remove', family, 'synthetic-package'],
                input=stubs + program, env=env, text=True, capture_output=True, timeout=10)

    def test_lifecycle_remove_rejects_active_services_and_fabric_avc(self):
        program = Path(__file__).with_name('lifecycle.sh').read_text()
        for unit in ('fabrico11y-server.service', 'fabrico11y-node.service'):
            for state in ('active', 'activating', 'deactivating', 'reloading'):
                with self.subTest(unit=unit, state=state):
                    result = self.shell(program, unit=unit, state=state)
                    self.assertEqual(result.returncode, 1, result.stdout + result.stderr)
                    self.assertNotIn('ACCEPT A13a PASS', result.stdout)
        for comm in ('fabric-server', 'fabric-node', 'fabricctl'):
            with self.subTest(comm=comm):
                result = self.shell(program, avc=f'type=AVC msg=audit(1.0:1): avc: denied {{ read }} comm="{comm}"')
                self.assertEqual(result.returncode, 1, result.stdout + result.stderr)
                self.assertNotIn('ACCEPT F4 PASS', result.stdout)
        for family in ('debian', 'fedora'):
            with self.subTest(clean=family):
                result = self.shell(program, family)
                self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
                self.assertIn('ACCEPT A13a PASS', result.stdout)
                self.assertIn('ACCEPT A13b PASS', result.stdout)

    def test_lifecycle_remove_rejects_missing_state_and_incomplete_purge(self):
        program = Path(__file__).with_name('lifecycle.sh').read_text()
        for kwargs, forbidden in (({'retained': False}, 'ACCEPT A13a PASS'),
                                  ({'purge_keeps_state': True}, 'ACCEPT A13b PASS')):
            with self.subTest(kwargs=kwargs):
                result = self.shell(program, 'debian', **kwargs)
                self.assertEqual(result.returncode, 1, result.stdout + result.stderr)
                self.assertNotIn(forbidden, result.stdout)

    def test_lifecycle_remove_rejects_unobserved_services_and_unreadable_audit(self):
        program = Path(__file__).with_name('lifecycle.sh').read_text()
        for mode in ('empty', 'incomplete', 'error'):
            with self.subTest(systemctl_mode=mode):
                result = self.shell(program, systemctl_mode=mode)
                self.assertEqual(result.returncode, 1, result.stdout + result.stderr)
                self.assertNotIn('ACCEPT A13a PASS', result.stdout)
        result = self.shell(program, grep_error=True)
        self.assertIn('synthetic audit read error', result.stderr)
        self.assertEqual(result.returncode, 1, result.stdout + result.stderr)
        self.assertNotIn('ACCEPT F4 PASS', result.stdout)

    def test_acceptance_remove_rejects_either_live_unit(self):
        source = Path(__file__).with_name('acceptance.sh').read_text()
        functions = '\n'.join(line for line in source.splitlines() if line.startswith(('pass()', 'fail()')))
        start = source.index('package_remove >')
        block = source[start:source.index('package_purge >', start)]
        for decision in ('stopped=$(systemctl is-active ', 'pass A13a ', 'fail A13a '):
            self.assertIn(decision, block, 'actual nonempty A13a decision required')
        program = 'set -u\nFAILS=0\nW=/root/accept\n' + functions + '\n' + block + '\nexit "$FAILS"\n'
        for unit in ('fabrico11y-server.service', 'fabrico11y-node.service'):
            for state in ('active', 'activating', 'deactivating', 'reloading', 'inactive'):
                with self.subTest(unit=unit, state=state):
                    result = self.shell(program, unit=unit, state=state)
                    self.assertEqual(result.returncode, 0 if state == 'inactive' else 1,
                                     result.stdout + result.stderr)
                    self.assertIn('ACCEPT A13a ' + ('PASS' if state == 'inactive' else 'FAIL'), result.stdout)
        for mode in ('empty', 'incomplete', 'error'):
            with self.subTest(systemctl_mode=mode):
                result = self.shell(program, systemctl_mode=mode)
                self.assertEqual(result.returncode, 1, result.stdout + result.stderr)
                self.assertNotIn('ACCEPT A13a PASS', result.stdout)

    def test_acceptance_fedora_rejects_unreadable_audit(self):
        source = Path(__file__).with_name('acceptance.sh').read_text()
        functions = '\n'.join(line for line in source.splitlines() if line.startswith(('pass()', 'fail()')))
        start = source.index('  audit_status=$(auditctl -s 2>&1)')
        block = source[start:source.index('\nfi\necho "RESULT', start)]
        for decision in ('ausearch -m AVC,USER_AVC ', 'selinux-avc.log', 'pass F3 ', 'fail F3 '):
            self.assertIn(decision, block, 'actual nonempty F3 audit decision required')
        program = 'set -u\nFAILS=0\nW=/root/accept\n' + functions + '\n' + block + '\nexit "$FAILS"\n'
        for grep_error in (False, True):
            with self.subTest(grep_error=grep_error):
                result = self.shell(program, grep_error=grep_error)
                if grep_error:
                    self.assertIn('synthetic audit read error', result.stderr)
                    self.assertEqual(result.returncode, 1, result.stdout + result.stderr)
                    self.assertNotIn('ACCEPT F3 PASS', result.stdout)
                else:
                    self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
                    self.assertIn('ACCEPT F3 PASS', result.stdout)


if __name__ == '__main__':
    unittest.main()