"""Bounded initial-thread syscall observer for the separately frozen helper."""
import ast
import hashlib
import json
import os
from pathlib import Path
import re
import select
import signal
import stat
import subprocess
import time

POPEN = subprocess.Popen
MAX_TRACE = 32 * 2**20
SYSCALLS = 'execve,execveat,open,openat,write,writev,fchmod,fsync,fdatasync,link,linkat,unlink,unlinkat,clone,clone3,fork,vfork,fstat,newfstatat,statx,close,close_range,dup,dup2,dup3,fcntl'
QUOTED = re.compile(r'"(?:[^"\\]|\\.)*"')


def require(value, reason):
    if not value:
        raise ValueError('native-trace-' + reason)


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


def strings(arguments):
    return [ast.literal_eval(value) for value in QUOTED.findall(arguments)]


def trace_calls(raw):
    require(type(raw) is bytes and len(raw) <= MAX_TRACE, 'bound')
    calls, offset = [], 0
    for line in raw.decode('ascii').splitlines(keepends=True):
        text = line.strip()
        match = re.fullmatch(r'([a-z0-9_]+)\((.*)\)\s+=\s+(-?(?:0x[0-9a-f]+|\d+))(.*)', text)
        if match:
            calls.append(dict(name=match[1], args=match[2], result=int(match[3], 0), tail=match[4],
                              offset=offset, line=text))
        else:
            require(not text or text.startswith(('--- ', '+++ ')), 'incomplete-or-unknown-line')
        offset += len(line.encode('ascii'))
    return calls


def publication(raw, begin_path, begin_raw, inode):
    """Facts from one ordered trace, plus independently observed final bytes/inode."""
    begin_path = Path(begin_path)
    value = json.loads(begin_raw)
    pid = value['process']['pid']
    stage = str(begin_path.parent / f'.begin-witness-{pid}.stage')
    calls = trace_calls(raw)
    execution = [call for call in calls if call['name'] in ('execve', 'execveat')]
    require(len(execution) == 1 and execution[0]['result'] == 0
            and execution[0]['name'] == 'execve' and '(DELAYED)' in execution[0]['tail'], 'exec')
    require(not any(call['name'] in ('fork', 'vfork') for call in calls), 'fork-escape')
    opened = [call for call in calls if call['name'] in ('open', 'openat') and stage in strings(call['args'])]
    require(len(opened) == 1 and opened[0]['result'] >= 0
            and all(flag in opened[0]['args'] for flag in ('O_RDWR', 'O_CREAT', 'O_EXCL', 'O_NOFOLLOW'))
            and re.search(r', 0600$', opened[0]['args']), 'exclusive-stage')
    descriptor = opened[0]['result']
    prefix = f'{descriptor}<{stage}>'
    def raw_write(call):
        if call['name'] not in ('write', 'writev'): return None
        match = re.fullmatch(r'(0x[0-9a-f]+|\d+), (0x[0-9a-f]+|\d+), (0x[0-9a-f]+|\d+)', call['args'])
        require(match is not None, 'redacted-write-format')
        return tuple(int(part, 0) for part in match.groups())
    def singular(name, predicate):
        selected = [call for call in calls if call['name'] in name and predicate(call)]
        require(len(selected) == 1 and selected[0]['result'] >= 0, 'operation-' + '-'.join(name))
        return selected[0]
    closing = [call for call in calls if call['name'] == 'close' and call['offset'] > opened[0]['offset']
               and re.match(r'^' + str(descriptor) + r'(?:<|$)', call['args'])]
    lifetime_end = closing[0]['offset'] if closing else len(raw)
    writes = [call for call in calls if call['name'] in ('write', 'writev')
              and opened[0]['offset'] < call['offset'] < lifetime_end and raw_write(call)[0] == descriptor]
    require(len(writes) == 1 and writes[0]['name'] == 'write', 'sole-stage-write')
    write = writes[0]
    require(write['result'] == raw_write(write)[2] == len(begin_raw), 'complete-write-count')
    require(not any(opened[0]['offset'] < call['offset'] < lifetime_end
                    and (call['name'] in ('dup', 'dup2', 'dup3', 'close_range')
                         or (call['name'] in ('open', 'openat') and call['result'] == descriptor)
                         or (call['name'] == 'fcntl' and call['args'].startswith(prefix) and 'F_DUPFD' in call['args']))
                    for call in calls), 'stage-fd-reuse')
    mode = singular(('fchmod',), lambda call: call['args'].startswith(prefix + ', '))
    require(mode['result'] == 0 and mode['args'].endswith(', 0400'), 'mode')
    synced = singular(('fsync',), lambda call: call['args'] == prefix)
    linked = singular(('link', 'linkat'), lambda call: strings(call['args']) == [stage, str(begin_path)])
    require(linked['result'] == 0 and (linked['name'] != 'linkat' or linked['args'].endswith(', 0')), 'no-replace-link')
    unlinked = singular(('unlink', 'unlinkat'), lambda call: strings(call['args']) == [stage])
    require(unlinked['result'] == 0, 'stage-unlink')
    directory = singular(('fsync',), lambda call: re.fullmatch(r'\d+<' + re.escape(str(begin_path.parent)) + '>', call['args']) is not None)
    require(directory['result'] == 0, 'directory-sync')
    steps = [opened[0], write, mode, synced, linked, unlinked, directory]
    require(all(a['offset'] < b['offset'] for a, b in zip(steps, steps[1:])), 'publication-order')
    clones = [call for call in calls if call['name'] in ('clone', 'clone3')]
    require(clones and all(call['offset'] > directory['offset'] for call in clones), 'before-worker')
    metadata = [call for call in calls if call['name'] in ('fstat', 'newfstatat', 'statx')
                and call['result'] == 0 and unlinked['offset'] < call['offset'] < clones[0]['offset']
                and (stage in call['args'] or str(begin_path) in strings(call['args']))
                and re.search(r'st(?:x)?_ino=' + str(inode) + r'(?:,|})', call['args'])
                and re.search(r'st(?:x)?_nlink=1(?:,|})', call['args'])
                and 'S_IFREG|0400' in call['args']]
    require(metadata, 'final-inode-metadata')
    require(lifetime_end > metadata[0]['offset'], 'stage-held-through-publication')
    return dict(atomic=True, exclusive=True, before_first_offer=True, regular=True,
                nlink=1, mode=0o400, unchanged_sha256=hashlib.sha256(begin_raw).hexdigest(),
                source_mapping='offers only in workers spawned after synchronous publication',
                operation_offsets={call['name'] + ':' + str(index): call['offset'] for index, call in enumerate(steps)},
                first_clone_offset=clones[0]['offset'], metadata_offset=metadata[0]['offset'])


class NativeTrace:
    @classmethod
    def start(cls, command, binary_fd, out, *, strace, strace_sha256, deadline,
              stdout=None, stderr=None, injection=None):
        self = cls()
        self.out, self.deadline = Path(out), deadline
        self.binary_fd, self.target_pidfd = os.dup(binary_fd), None
        self.files, self.target_pid, self.returncode = [], None, None
        self.tracer = None
        self.record = dict(state='starting', command=command, bootstrap_delay='first execve delay_exit=100ms')
        try:
            require(self.out.is_absolute() and not self.out.exists()
                    and self.out.parent.resolve().is_relative_to(Path('/run/media/kmosoti/data/FabricO11y/results')), 'output')
            self.out.mkdir(mode=0o700)
            require(not Path(strace).is_symlink() and sha(strace) == strace_sha256, 'tool-pin')
            original = os.fstat(self.binary_fd)
            self.binary_sha256 = sha(Path('/proc/self/fd') / str(self.binary_fd))
            require(command[0] == f'/proc/self/fd/{binary_fd}', 'held-exec-command')
            require(not any(name in os.environ for name in ('LD_PRELOAD', 'LD_AUDIT', 'LD_LIBRARY_PATH')), 'environment')
            self.trace = self.out / 'syscalls.raw'
            for name, stream in (('stdout', stdout), ('stderr', stderr)):
                if stream is None:
                    stream = (self.out / name).open('xb'); self.files.append(stream)
                setattr(self, name, stream)
            argv = [str(strace), '-q', '-T', '-v', '-x', '-yy', '-s', '131072', '-o', str(self.trace),
                    '-e', 'trace=' + SYSCALLS, '-e', 'raw=write,writev', '-e', 'inject=execve:delay_exit=100ms:when=1']
            if injection is not None:
                require(re.fullmatch(r'(write|fsync|link|linkat|unlink|unlinkat):error=EIO:when=[1-9][0-9]*', injection), 'injection')
                argv += ['-e', 'inject=' + injection]
            argv += ['--', *command]
            self.record.update(tracer_command=argv, tool_sha256=strace_sha256, launched_monotonic_ns=time.monotonic_ns())
            environment = {'PATH': '/usr/bin:/bin', 'LANG': 'C', 'LC_ALL': 'C'}
            self.record['environment'] = environment
            self.tracer = POPEN(argv, pass_fds=(binary_fd,), stdout=self.stdout, stderr=self.stderr, env=environment)
            until = min(deadline, time.monotonic() + 5)
            while time.monotonic() < until and self.tracer.poll() is None:
                children = Path(f'/proc/{self.tracer.pid}/task/{self.tracer.pid}/children').read_text().split()
                require(len(children) <= 1, 'tracer-child-domain')
                if children:
                    pid = int(children[0]); proc = Path('/proc') / str(pid)
                    try:
                        first = (proc / 'stat').read_text().rsplit(')', 1)[1].split()
                        loaded = (proc / 'exe').stat()
                        if (loaded.st_dev, loaded.st_ino) == (original.st_dev, original.st_ino):
                            descriptor = os.pidfd_open(pid)
                            self.target_pidfd = descriptor
                            actual_sha = sha(proc / 'exe')
                            last = (proc / 'stat').read_text().rsplit(')', 1)[1].split()
                            require(first[19] == last[19] and actual_sha == self.binary_sha256
                                    and last[0] == 't' and len(list((proc / 'task').iterdir())) == 1, 'bootstrap-binding')
                            self.target_pid, self.target_pidfd = pid, descriptor
                            self.identity = dict(pid=pid, start_ticks=int(last[19]),
                                boot_id=Path('/proc/sys/kernel/random/boot_id').read_text().strip(),
                                exe_device=loaded.st_dev, exe_inode=loaded.st_ino)
                            self.record.update(state='bound', process=self.identity, binary_sha256=self.binary_sha256,
                                bound_monotonic_ns=time.monotonic_ns(), bootstrap_state=last[0], tracer_pid=self.tracer.pid)
                            return self
                    except FileNotFoundError:
                        pass
                time.sleep(.001)
            raise ValueError('native-trace-live-binding-unavailable')
        except BaseException:
            self.abort()
            raise

    @property
    def pid(self):
        return self.target_pid

    def poll(self):
        if self.trace.exists(): require(self.trace.stat().st_size <= MAX_TRACE, 'bound')
        if self.returncode is not None: return self.returncode
        if self.tracer.poll() is None: return None
        raw = self.trace.read_bytes()
        status = re.findall(rb'^\+\+\+ exited with ([0-9]+) \+\+\+\s*$', raw, re.M)
        killed = re.findall(rb'^\+\+\+ killed by (SIG[A-Z0-9]+).*$', raw, re.M)
        require(len(status) + len(killed) == 1, 'target-exit')
        self.returncode = int(status[0]) if status else -int(getattr(signal, killed[0].decode()))
        return self.returncode

    def wait(self, timeout=None):
        until = min(self.deadline, time.monotonic() + timeout if timeout is not None else self.deadline)
        while self.poll() is None:
            if time.monotonic() >= until: raise subprocess.TimeoutExpired(self.record['command'], timeout)
            time.sleep(.005)
        self.tracer.wait(timeout=max(0, until-time.monotonic()))
        return self.returncode

    def send_signal(self, number):
        require(self.target_pidfd is not None, 'unbound-signal')
        signal.pidfd_send_signal(self.target_pidfd, number)

    def kill(self):
        self.send_signal(signal.SIGKILL)

    def finish(self, beginfile):
        try:
            return self._finish(beginfile)
        except BaseException as error:
            self.record['error'] = type(error).__name__ + ': ' + str(error)
            self.abort()
            raise

    def _finish(self, beginfile):
        require(self.wait(timeout=max(0, self.deadline-time.monotonic())) == 0, 'successful-helper')
        require(self.tracer.returncode == 0, 'successful-tracer')
        beginfile = Path(beginfile)
        info = beginfile.lstat()
        require(stat.S_ISREG(info.st_mode) and info.st_nlink == 1 and stat.S_IMODE(info.st_mode) == 0o400
                and info.st_size <= 131072, 'begin-file')
        raw = beginfile.read_bytes()
        value = json.loads(raw)
        require(value['process'] == self.identity and value['binary_sha256'] == self.binary_sha256, 'begin-identity')
        facts = publication(self.trace.read_bytes(), beginfile, raw, info.st_ino)
        self.record.update(state='stopped', exit=self.returncode, tracer_exit=self.tracer.returncode,
            trace_sha256=sha(self.trace), begin_sha256=hashlib.sha256(raw).hexdigest(), begin_inode=info.st_ino,
            begin_path=str(beginfile),
            publication=facts, stopped_monotonic_ns=time.monotonic_ns())
        self.archive()
        self.release()
        return self.record

    def archive(self):
        if self.out.exists(): (self.out / 'observer-receipt.json').write_text(json.dumps(self.record, indent=2) + '\n')

    def release(self):
        for name in ('binary_fd', 'target_pidfd'):
            descriptor = getattr(self, name, None)
            if descriptor is not None: os.close(descriptor); setattr(self, name, None)
        for stream in self.files: stream.close()
        self.files = []

    def abort(self):
        try:
            self.record['outer_campaign_stop_required'] = self.tracer is not None and self.target_pidfd is None
            if self.target_pidfd is not None:
                try: signal.pidfd_send_signal(self.target_pidfd, signal.SIGKILL)
                except ProcessLookupError: pass
            if self.tracer is not None:
                try: self.tracer.wait(timeout=5)
                except subprocess.TimeoutExpired:
                    self.tracer.kill(); self.tracer.wait(timeout=5)
            self.record.update(state='failed', tracer_exit=None if self.tracer is None else self.tracer.returncode)
        finally:
            try: self.archive()
            finally: self.release()

    @staticmethod
    def validate_stopped(rawtrace_dir, beginfile, expected):
        directory, beginfile = Path(rawtrace_dir), Path(beginfile)
        receipt = directory / 'observer-receipt.json'
        trace = directory / 'syscalls.raw'
        require(sha(receipt) == expected['observer_receipt_sha256'] and sha(trace) == expected['trace_sha256']
                and sha(beginfile) == expected['begin_sha256'], 'offline-expected-pins')
        record = json.loads(receipt.read_text())
        require(record['state'] == 'stopped' and record['exit'] == record['tracer_exit'] == 0
                and record['process'] == expected['process']
                and record['binary_sha256'] == expected['binary_sha256'], 'offline-identity')
        require(beginfile.stat().st_ino == record['begin_inode'], 'offline-inode')
        facts = publication(trace.read_bytes(), Path(record['begin_path']), beginfile.read_bytes(), record['begin_inode'])
        require(facts == record['publication'], 'offline-publication')
        return facts

"""Explicit finite publication controls; uses the original TLS diagnostic fixture."""
import argparse
import hashlib
import json
import os
from pathlib import Path
import re
import signal
import sys
import threading
import time
from unittest.mock import patch

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from resource_group import require_limits, require_storage
import simulator_native_controls as native
from simulator_native_trace import NativeTrace, POPEN, publication, require, sha, strings, trace_calls


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--case', choices=('positive', 'write', 'sync', 'link', 'unlink', 'existing', 'stage', 'stage-symlink'), required=True)
    for option in ('binary', 'authority', 'build-receipt', 'strace', 'out'):
        parser.add_argument('--' + option, type=Path, required=True)
    for option in ('authority-sha256', 'strace-sha256'):
        parser.add_argument('--' + option, required=True)
    parser.add_argument('--calibration', type=Path)
    parser.add_argument('--calibration-sha256')
    args = parser.parse_args()
    require_limits()
    data = require_storage()
    require(args.out.parent.resolve().is_relative_to(data / 'results') and not args.out.exists(), 'control-output')
    args.out.mkdir(mode=0o700)
    report = dict(case=args.case, classification='finite native syscall publication control only', exit=2,
                  authority_sha256=args.authority_sha256, tool_sha256=args.strace_sha256)
    handles, sinks, victim, selected = [], [], None, None
    old_argv = sys.argv
    original_case = 'existing' if args.case == 'existing' else 'positive'
    fixture = args.out / 'fixture'
    deadline = time.monotonic() + 120
    previous_alarm = signal.signal(signal.SIGALRM, lambda *_: (_ for _ in ()).throw(TimeoutError('native trace controller deadline')))
    signal.alarm(120)
    class TrackedSink(native.Sink):
        loop_thread = None
        def __init__(self, *values, **options):
            super().__init__(*values, **options)
            sinks.append(self)
        def serve_forever(self, *values, **options):
            self.loop_thread = threading.current_thread()
            return super().serve_forever(*values, **options)
    try:
        if args.case in ('write', 'sync', 'link', 'unlink'):
            require(args.calibration is not None and sha(args.calibration) == args.calibration_sha256, 'calibration-pin')
            calibration = json.loads(args.calibration.read_text())
            require(calibration['authority_sha256'] == args.authority_sha256
                    and calibration['tool_sha256'] == args.strace_sha256
                    and calibration['binary_sha256'] == sha(args.binary), 'calibration-inputs')
            require(sha(Path(calibration['positive_trace'])) == calibration['positive_trace_sha256']
                    and sha(Path(calibration['positive_observer_receipt'])) == calibration['positive_observer_receipt_sha256'], 'calibration-evidence')
            selected = calibration['selectors'][args.case]
        def start(command, *positional, **options):
            nonlocal victim
            if not (isinstance(command, list) and command and command[0].startswith('/proc/self/fd/')
                    and '--server-url' in command):
                return POPEN(command, *positional, **options)
            require(not handles and not positional, 'one-target')
            fd = int(command[0].rsplit('/', 1)[1])
            require(options['pass_fds'] == (fd,), 'held-target-fd')
            injection = None if selected is None else f"{selected['syscall']}:error=EIO:when={selected['ordinal']}"
            handle = NativeTrace.start(command, fd, args.out / 'trace', strace=args.strace,
                strace_sha256=args.strace_sha256, deadline=deadline, stdout=options['stdout'],
                stderr=options['stderr'], injection=injection)
            handles.append(handle)
            if args.case in ('stage', 'stage-symlink'):
                stage = fixture / f'.begin-witness-{handle.pid}.stage'
                victim = fixture / 'preserved-victim'
                victim.write_bytes(b'preserve exact owned victim\n')
                victim.chmod(0o400)
                if args.case == 'stage-symlink': stage.symlink_to(victim)
                else: stage.write_bytes(b'preserve exact owned stage\n')
            return handle
        sys.argv = [native.__file__, '--case', original_case, '--binary', str(args.binary),
            '--authority', str(args.authority), '--authority-sha256', args.authority_sha256,
            '--build-receipt', str(args.build_receipt), '--out', str(fixture)]
        with patch.object(native.subprocess, 'Popen', start), patch.object(native, 'Sink', TrackedSink):
            fixture_exit = native.main()
        require(len(handles) == 1, 'actual-target')
        handle = handles[0]
        target_exit = handle.wait(timeout=max(0, deadline-time.monotonic()))
        original = json.loads((fixture / 'control-receipt.json').read_text())
        require(original['helper_stopped'] and original['sink_stopped'], 'fixture-stop')
        report.update(fixture_exit=fixture_exit, target_exit=target_exit, tracer_exit=handle.tracer.returncode,
                      process=handle.identity, original_fixture_receipt_sha256=sha(fixture / 'control-receipt.json'))
        raw = handle.trace.read_bytes()
        calls = trace_calls(raw)
        stage = str(fixture / f'.begin-witness-{handle.pid}.stage')
        begin = fixture / 'begin.json'
        if args.case == 'positive':
            require(fixture_exit == target_exit == 0, 'positive-fixture')
            observer = handle.finish(begin)
            offsets = observer['publication']['operation_offsets']
            selectors = {}
            for key, step in (('write', 1), ('sync', 3), ('link', 4), ('unlink', 5)):
                offset = next(value for name, value in offsets.items() if name.endswith(':' + str(step)))
                call = next(call for call in calls if call['offset'] == offset)
                selectors[key] = dict(syscall=call['name'], ordinal=sum(
                    item['name'] == call['name'] and item['offset'] <= offset for item in calls), positive_offset=offset)
            calibration = dict(schema='native-publication-calibration-v1', selectors=selectors,
                authority_sha256=args.authority_sha256, binary_sha256=sha(args.binary), tool_sha256=args.strace_sha256,
                positive_trace=str(handle.trace), positive_trace_sha256=sha(handle.trace),
                positive_observer_receipt=str(args.out / 'trace/observer-receipt.json'),
                positive_observer_receipt_sha256=sha(args.out / 'trace/observer-receipt.json'))
            (args.out / 'calibration.json').write_text(json.dumps(calibration, indent=2) + '\n')
            report.update(publication=observer['publication'], calibration_sha256=sha(args.out / 'calibration.json'))
        else:
            require(target_exit == 1 and not any(call['name'] in ('clone', 'clone3', 'fork', 'vfork') for call in calls), 'refusal-before-worker')
            if selected is not None:
                matches = [call for call in calls if '(INJECTED)' in call['tail']]
                require(len(matches) == 1, 'one-injected-operation')
                fault = matches[0]
                require(fault['name'] == selected['syscall'] and fault['result'] == -1 and 'EIO' in fault['tail']
                    and sum(call['name'] == fault['name'] and call['offset'] <= fault['offset'] for call in calls) == selected['ordinal'], 'actual-injected-operation')
                if args.case in ('write', 'sync'):
                    if args.case == 'write':
                        opened = [call for call in calls if call['name'] in ('open', 'openat')
                                  and stage in strings(call['args']) and call['result'] >= 0]
                        require(len(opened) == 1 and int(fault['args'].split(',', 1)[0], 0) == opened[0]['result'], 'injected-stage-fd')
                    else:
                        require(re.match(r'\d+<' + re.escape(stage) + r'>', fault['args']), 'injected-stage-fd')
                else:
                    require(strings(fault['args']) == ([stage, str(begin)] if args.case == 'link' else [stage]), 'injected-publication-path')
                report['injected_operation'] = fault
            elif args.case == 'existing':
                require(fixture_exit == 0 and begin.read_bytes() == b'never replace this destination\n'
                        and any(call['name'] in ('link', 'linkat') and strings(call['args']) == [stage, str(begin)]
                                and call['result'] == -1 and 'EEXIST' in call['tail'] for call in calls), 'existing-destination')
            else:
                require(any(call['name'] in ('open', 'openat') and stage in strings(call['args'])
                            and call['result'] == -1 and 'EEXIST' in call['tail'] for call in calls), 'stage-collision')
                require(victim.read_bytes() == b'preserve exact owned victim\n', 'victim-preserved')
                if args.case == 'stage': require(Path(stage).read_bytes() == b'preserve exact owned stage\n', 'stage-preserved')
            require(not any(fixture.glob('batch-*')), 'no-wire-offer')
            handle.record.update(state='expected-publication-refusal', exit=target_exit,
                tracer_exit=handle.tracer.returncode, trace_sha256=sha(handle.trace))
            handle.archive(); handle.release()
        report.update(exit=0, status='finite-control-satisfied', trace_sha256=sha(args.out / 'trace/syscalls.raw'))
    except BaseException as error:
        report['error'] = type(error).__name__ + ': ' + str(error)
    finally:
        sys.argv = old_argv
        for handle in handles:
            if handle.binary_fd is not None: handle.abort()
        for sink in sinks:
            if sink.loop_thread is not None and sink.loop_thread.is_alive():
                sink.shutdown(); sink.loop_thread.join(timeout=5)
            sink.server_close()
            if sink.loop_thread is not None and sink.loop_thread.is_alive():
                report.update(exit=2, cleanup_error='owned TLS sink not stopped')
        signal.alarm(0)
        signal.signal(signal.SIGALRM, previous_alarm)
        (args.out / 'receipt.json').write_text(json.dumps(report, indent=2) + '\n')
    return report['exit']


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

"""Literal syscall transcripts; no native trace behavior is inferred."""
import json
import unittest
from pathlib import Path

import simulator_native_trace as trace

BEGIN = b'{"process":{"pid":7}}'
PATH = Path('/owned/begin.json')
STAGE = '/owned/.begin-witness-7.stage'
LINES = [
    'execve("/proc/self/fd/9", [], 0) = 0 (DELAYED) <0.100001>',
    f'openat(AT_FDCWD, "{STAGE}", O_RDWR|O_CREAT|O_EXCL|O_NOFOLLOW|O_CLOEXEC, 0600) = 3<{STAGE}>',
    f'write(0x3, 0x1234, {hex(len(BEGIN))}) = {hex(len(BEGIN))}',
    f'fchmod(3<{STAGE}>, 0400) = 0',
    f'fsync(3<{STAGE}>) = 0',
    f'linkat(AT_FDCWD, "{STAGE}", AT_FDCWD, "/owned/begin.json", 0) = 0',
    f'unlink("{STAGE}") = 0',
    'fsync(4</owned>) = 0',
    f'fstat(3<{STAGE} (deleted)>, {{st_mode=S_IFREG|0400, st_ino=99, st_nlink=1, st_size={len(BEGIN)}}}) = 0',
    'clone3({flags=CLONE_VM|CLONE_THREAD}, 88) = 8',
    'exit_group(0) = ?',
    '+++ exited with 0 +++',
]


def fixture(lines=None):
    # exit_group normally has no numeric return and is not a publication step.
    return ('\n'.join(LINES[:-2] + LINES[-1:] if lines is None else lines) + '\n').encode()


class NativeTraceLiteralTests(unittest.TestCase):
    def test_complete_literal_sequence(self):
        facts = trace.publication(fixture(), PATH, BEGIN, 99)
        self.assertEqual((facts['atomic'], facts['exclusive'], facts['before_first_offer']), (True, True, True))

    def test_missing_each_publication_step(self):
        for index in range(1, 9):
            with self.subTest(index=index), self.assertRaises(ValueError):
                trace.publication(fixture(LINES[:index] + LINES[index+1:-2] + LINES[-1:]), PATH, BEGIN, 99)

    def test_reordered_sync_link_unlink(self):
        for a, b in ((3, 4), (4, 5), (5, 6), (6, 7)):
            lines = list(LINES[:-2] + LINES[-1:]); lines[a], lines[b] = lines[b], lines[a]
            with self.subTest(a=a, b=b), self.assertRaises(ValueError):
                trace.publication(fixture(lines), PATH, BEGIN, 99)

    def test_short_write_and_substituted_count(self):
        for replacement in (LINES[2].rsplit(' = ', 1)[0] + ' = 0x1', LINES[2].replace(hex(len(BEGIN)) + ')', '0xff)')):
            lines = list(LINES[:-2] + LINES[-1:]); lines[2] = replacement
            with self.subTest(replacement=replacement), self.assertRaises(ValueError):
                trace.publication(fixture(lines), PATH, BEGIN, 99)

    def test_exclusive_nofollow_and_mode(self):
        for index, old, new in ((1, 'O_EXCL|', ''), (1, 'O_NOFOLLOW|', ''), (3, '0400', '0600')):
            lines = list(LINES[:-2] + LINES[-1:]); lines[index] = lines[index].replace(old, new)
            with self.subTest(old=old), self.assertRaises(ValueError):
                trace.publication(fixture(lines), PATH, BEGIN, 99)

    def test_wrong_destination_parent_and_inode(self):
        for index, old, new in ((5, '/owned/begin.json', '/owned/other.json'), (7, '4</owned>', '4</foreign>'), (8, 'st_ino=99', 'st_ino=100'), (8, 'st_nlink=1', 'st_nlink=2')):
            lines = list(LINES[:-2] + LINES[-1:]); lines[index] = lines[index].replace(old, new)
            with self.subTest(old=old), self.assertRaises(ValueError):
                trace.publication(fixture(lines), PATH, BEGIN, 99)

    def test_clone_before_publication_or_no_clone(self):
        for lines in (LINES[:1] + [LINES[9]] + LINES[1:9] + LINES[-1:], LINES[:9] + LINES[-1:]):
            with self.subTest(lines=lines), self.assertRaises(ValueError):
                trace.publication(fixture(lines), PATH, BEGIN, 99)

    def test_failed_syscalls_and_fork_escape(self):
        for index in range(1, 8):
            lines = list(LINES[:-2] + LINES[-1:]); lines[index] = lines[index].split(' = ')[0] + ' = -1 EIO (Input/output error) (INJECTED)'
            with self.subTest(index=index), self.assertRaises(ValueError):
                trace.publication(fixture(lines), PATH, BEGIN, 99)
        with self.assertRaisesRegex(ValueError, 'fork-escape'):
            trace.publication(fixture(LINES[:-2] + ['fork() = 12'] + LINES[-1:]), PATH, BEGIN, 99)

    def test_missing_delayed_exec_or_multiple_exec(self):
        for lines in ([LINES[0].replace(' (DELAYED)', '')] + LINES[1:-2] + LINES[-1:], [LINES[0]] + LINES[:-2] + LINES[-1:]):
            with self.subTest(lines=lines), self.assertRaises(ValueError):
                trace.publication(fixture(lines), PATH, BEGIN, 99)

    def test_incomplete_or_oversized_trace_refuses(self):
        with self.assertRaises(ValueError): trace.trace_calls(b'write(3, "x", 1 <unfinished ...>\n')
        with self.assertRaises(ValueError): trace.trace_calls(b'x' * (trace.MAX_TRACE + 1))