import datetime, hashlib, json, os, re, signal, subprocess, sys, time
from pathlib import Path
ROOT = Path('/run/media/kmosoti/data/FabricO11y')
REPO = ROOT / 'scratch/installation-custody-witness'
OUT = ROOT / 'results/service-opening-candidate'
STORE = ROOT / 'scratch/service-opening-census'
FROZEN = 'be1fb749c15b11d4f9dffc35edd75755511c0b66'
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)

"""Packaging regressions: exact assets, tamper rejection and bounded unit policy."""
import ast
import hashlib
import io
import json
import lzma
import os
from pathlib import Path
import shutil
import signal
import subprocess
import sys
import tempfile
import tarfile
import time
from types import SimpleNamespace
import unittest
import zlib
from unittest.mock import patch

import dependency_notices
import build_candidate_rpm
from test_payload import assert_fixture_stopped, owned_process_fixture

from stage_console import stage
from stage_predecessor import stage_predecessor
from stage_candidate import stage_candidate, validate_control

ROOT = Path(__file__).resolve().parents[2]


def png(extra=None):
    """A minimal valid 1x1 PNG, optionally with one metadata chunk before IEND."""
    def chunk(kind, data):
        return len(data).to_bytes(4, 'big') + kind + data + zlib.crc32(kind + data).to_bytes(4, 'big')
    body = b'\x89PNG\r\n\x1a\n' + chunk(b'IHDR', bytes([0, 0, 0, 1, 0, 0, 0, 1, 8, 0, 0, 0, 0]))
    body += chunk(b'IDAT', zlib.compress(b'\x00\x00'))
    if extra == b'tIME':
        body += chunk(extra, bytes([7, 234, 10, 12, 17, 0, 6]))
    elif extra:
        body += chunk(extra, b'date:modify\x002026-10-12T10:00:00+00:00')
    return body + chunk(b'IEND', b'')


class PackageTests(unittest.TestCase):
    def test_build_stops_real_descendant_on_success_and_timeout(self):
        # Only the test's real communicate timer is shortened; production requests 300s.
        real_popen = subprocess.Popen
        for mode in ('success', 'timeout'):
            with self.subTest(mode=mode), owned_process_fixture() as (directory, command):
                timeouts = []
                def spawn(*args, **kwargs):
                    process = real_popen(*args, **kwargs)
                    communicate = process.communicate
                    def bounded_communicate(*args, **kwargs):
                        timeouts.append(kwargs.get('timeout'))
                        if kwargs.get('timeout') == 300:
                            kwargs['timeout'] = 1
                        return communicate(*args, **kwargs)
                    process.communicate = bounded_communicate
                    return process
                with patch.object(build_candidate_rpm.subprocess, 'Popen', side_effect=spawn):
                    if mode == 'success':
                        self.assertEqual(build_candidate_rpm.run_build(command + [mode], os.environ.copy(), directory), 0)
                    else:
                        with self.assertRaises(subprocess.TimeoutExpired):
                            build_candidate_rpm.run_build(command + [mode], os.environ.copy(), directory)
                self.assertIn(300, timeouts, 'native build deadline changed in the control')
                self.assertEqual((directory / 'build.stdout').read_text(), 'literal process stdout\n')
                self.assertEqual((directory / 'build.stderr').read_text(), 'literal process stderr\n')
                assert_fixture_stopped(self, directory)

    def test_build_final_drain_interruption_preserves_raw_diagnostics(self):
        # Inject a typed final-drain error only AFTER a real bounded drain/reap.
        real_popen = subprocess.Popen
        with owned_process_fixture() as (directory, command):
            timeouts = []
            def spawn(*args, **kwargs):
                process = real_popen(*args, **kwargs)
                communicate = process.communicate
                def bounded_communicate(*args, **kwargs):
                    timeouts.append(kwargs.get('timeout'))
                    kwargs['timeout'] = 1
                    stdout, stderr = communicate(*args, **kwargs)
                    if len(timeouts) == 3:
                        raise subprocess.TimeoutExpired(process.args, 10, output=stdout, stderr=stderr)
                    return stdout, stderr
                process.communicate = bounded_communicate
                return process
            with patch.object(build_candidate_rpm.subprocess, 'Popen', side_effect=spawn):
                with self.assertRaises(subprocess.TimeoutExpired):
                    build_candidate_rpm.run_build(command + ['drain'], os.environ.copy(), directory)
            self.assertEqual(timeouts, [300, 10, 10])
            assert_fixture_stopped(self, directory)
            for stream in ('stdout', 'stderr'):
                with self.subTest(stream=stream):
                    path = directory / ('build.' + stream)
                    self.assertTrue(path.is_file(), 'final drain interruption lost raw ' + stream)
                    self.assertEqual(path.read_text(), 'literal process ' + stream + '\n')

    def test_constructor_parent_term_preserves_interrupted_receipt_and_stage(self):
        # Admission is synthetic; signals, children, staging and diagnostics are real.
        with owned_process_fixture() as (directory, command):
            driver = directory / 'constructor.py'
            driver.write_text('''import json, os, sys
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import patch
import build_candidate_rpm as builder
directory = Path(sys.argv[1])
controls = json.loads((builder.ROOT / 'docs/experiments/formal/fixtures/owner-walkthrough.json').read_text())
packages = {family: directory / family for family in ('debian', 'fedora')}
for package in packages.values(): package.write_bytes(b'literal package stand-in')
receipt = directory / 'input.json'
receipt.write_text(json.dumps({'source': {'original_head': controls['original_head'], 'source_date_epoch': controls['source_date_epoch']}}))
def command_output(command):
    if command == ['dpkg', '--compare-versions', '0.1.0~alpha.0', 'lt', '0.1.0~alpha.1']: return b''
    if command[0] == 'dpkg-deb': return b'0.1.0~alpha.1'
    if '--eval' in command: return b'-1'
    return b'0' if '%{EPOCHNUM}' in command else b'0.1.0-0.alpha.1'
real_build = builder.run_build
def owned_build(command, env, out, label):
    if command[:2] != ['dpkg-deb', '-R'] or label != 'deb-unpack': raise AssertionError('unexpected native build command')
    stage = Path(command[-1]).parent
    return real_build([sys.executable, '-B', str(directory / 'child.py'), str(directory), 'timeout', str(stage)], env, out, label)
sys.argv = ['constructor.py', '--same-code-fixture', '--deb-receipt', str(receipt), '--rpm-receipt', str(receipt),
            '--out', str(directory / 'results/out'), '--frozen', controls['original_head'],
            '--source-date-epoch', str(controls['source_date_epoch']), '--harness-commit', '1' * 40]
with patch.dict('sys.modules', {'inputs': SimpleNamespace(validate_inputs=lambda *args: packages)}), \\
     patch.object(builder, 'STORAGE', directory), \\
     patch.object(builder, 'validate_harness'), \\
     patch.object(builder, 'inventory', return_value={'usr/bin/fabric-server': {'bytes': 3}}), \\
     patch.object(builder, 'deb_controls', return_value={'preinst': {'bytes': 3}}), \\
     patch.object(builder, 'rpm_records', return_value={'scripts': {'PREIN': {'bytes': 3}}, 'metadata': {}}), \\
     patch.object(builder, 'bounded_command', side_effect=command_output), \\
     patch.object(builder, 'run_build', side_effect=owned_build):
    raise SystemExit(builder.main())
''')
            scratch = directory / 'scratch'
            scratch.mkdir()
            env = {**os.environ, 'PYTHONPATH': str(ROOT / 'tools/packaging'), 'FABRIC_SCRATCH_ROOT': str(scratch)}
            process = subprocess.Popen([sys.executable, '-B', str(driver), str(directory)], env=env,
                                       stdout=subprocess.PIPE, stderr=subprocess.PIPE, start_new_session=True)
            descriptor = os.pidfd_open(process.pid)
            try:
                deadline = time.monotonic() + 5
                while not (directory / 'diagnostics-ready').exists() and process.poll() is None and time.monotonic() < deadline:
                    time.sleep(.01)
                self.assertTrue((directory / 'diagnostics-ready').is_file(), 'constructor did not reach real owned staging')
                signal.pidfd_send_signal(descriptor, signal.SIGTERM)
                stdout, stderr = process.communicate(timeout=8)
                self.assertEqual(process.returncode, 2, stdout + stderr)
                path = directory / 'results/out/fixture-receipt.json'
                self.assertTrue(path.is_file(), 'parent TERM lost interrupted receipt; exit=' + str(process.returncode))
                report = json.loads(path.read_text())
                self.assertEqual((report['state'], report['exit']), ('interrupted', 2))
                retained = Path(report['failure_evidence'])
                self.assertTrue(retained.is_relative_to(directory / 'results/out'))
                self.assertEqual((retained / 'held-stage.bin').read_bytes(), b'literal owned staging')
                for stream in ('stdout', 'stderr'):
                    self.assertEqual((directory / ('results/out/deb-unpack.' + stream)).read_text(),
                                     'literal process ' + stream + '\n')
                self.assertEqual(list(scratch.iterdir()), [])
                assert_fixture_stopped(self, directory)
            finally:
                if process.poll() is None:
                    signal.pidfd_send_signal(descriptor, signal.SIGKILL)
                process.wait(timeout=3)
                os.close(descriptor)
                process.stdout.close()
                process.stderr.close()

    def test_constructor_closeout_term_preserves_receipt_and_stage(self):
        # Real main/TERM and extracted actual closeout; native admission is not exercised.
        program = '''import ast, json, shutil, sys, time
from pathlib import Path
from unittest.mock import patch
import build_candidate_rpm as builder
directory = Path(sys.argv[1])
out, work = directory / 'out', directory / 'stage'
out.mkdir(); work.mkdir()
(work / 'held-stage.bin').write_bytes(b'literal completed staging')
for stream in ('stdout', 'stderr'):
    (out / ('build.' + stream)).write_bytes(('literal closeout ' + stream + '\\n').encode())
function = next(node for node in ast.parse(Path(builder.__file__).read_text()).body
                if isinstance(node, ast.FunctionDef) and node.name == 'fixture_main')
final = next(node.finalbody for node in function.body if isinstance(node, ast.Try) and node.finalbody)
assert final and any('fixture-receipt.json' in ast.unparse(node) for node in final)
code = compile(ast.Module(body=final, type_ignores=[]), '<actual-constructor-closeout>', 'exec')
real_remove = shutil.rmtree
def remove(path, *args, **kwargs):
    assert Path(path) == work
    pending = directory / 'ready.pending'
    pending.write_text('actual completed-stage removal boundary')
    pending.replace(directory / 'ready')
    deadline = time.monotonic() + 5
    while not (directory / 'release').exists():
        if time.monotonic() >= deadline: raise RuntimeError('closeout gate timed out')
        time.sleep(.01)
    return real_remove(path, *args, **kwargs)
def closeout(args):
    namespace = dict(builder.__dict__, out=out, work=work, record={'state': 'built', 'exit': 0})
    exec(code, namespace)
    return namespace['record']['exit']
sys.argv = ['closeout.py', '--same-code-fixture', '--deb-receipt', str(directory / 'synthetic-input'), '--out', str(out)]
with patch.object(builder, 'fixture_main', closeout), patch.object(builder.shutil, 'rmtree', remove):
    raise SystemExit(builder.main())
'''
        for interrupted in (False, True):
            with self.subTest(interrupted=interrupted):
                directory = Path(tempfile.mkdtemp(dir=os.environ['FABRIC_SCRATCH_ROOT'], prefix='constructor-closeout-'))
                script = directory / 'closeout.py'; script.write_text(program)
                env = {**os.environ, 'PYTHONPATH': str(ROOT / 'tools/packaging')}
                process = subprocess.Popen([sys.executable, '-B', str(script), str(directory)], env=env,
                                           stdout=subprocess.PIPE, stderr=subprocess.PIPE, start_new_session=True)
                descriptor = os.pidfd_open(process.pid)
                stopped = False
                try:
                    deadline = time.monotonic() + 5
                    while not (directory / 'ready').exists() and process.poll() is None and time.monotonic() < deadline:
                        time.sleep(.01)
                    self.assertTrue((directory / 'ready').is_file(), 'actual closeout boundary was not reached')
                    if interrupted: signal.pidfd_send_signal(descriptor, signal.SIGTERM)
                    else: (directory / 'release').touch()
                    stdout, stderr = process.communicate(timeout=8)
                    self.assertEqual(process.returncode, 2 if interrupted else 0, stdout + stderr)
                    receipt = directory / 'out/fixture-receipt.json'
                    self.assertTrue(receipt.is_file(), 'closeout TERM lost the actual receipt after staging')
                    report = json.loads(receipt.read_text())
                    self.assertEqual((report['state'], report['exit']), ('interrupted', 2) if interrupted else ('built', 0))
                    if interrupted:
                        retained = Path(report['failure_evidence'])
                        self.assertTrue(retained.is_relative_to(directory))
                        self.assertEqual((retained / 'held-stage.bin').read_bytes(), b'literal completed staging')
                        self.assertIsNot(report.get('cleanup_confirmed'), True)
                    else: self.assertFalse((directory / 'stage').exists())
                    for stream in ('stdout', 'stderr'):
                        self.assertEqual((directory / ('out/build.' + stream)).read_bytes(),
                                         ('literal closeout ' + stream + '\n').encode())
                finally:
                    try:
                        if process.poll() is None: signal.pidfd_send_signal(descriptor, signal.SIGKILL)
                        process.wait(timeout=3); stopped = True
                    finally:
                        os.close(descriptor); process.stdout.close(); process.stderr.close()
                        if stopped: shutil.rmtree(directory)
                        else: raise AssertionError('unconfirmed closeout process; fixture retained at ' + str(directory))

    def test_same_code_fixture_archive_errors_preserve_failed_receipt(self):
        # Origin: #59 C-041 SF2, a corrupt archive escaped after staging began.
        controls = json.loads((ROOT / 'docs/experiments/formal/fixtures/owner-walkthrough.json').read_text())
        for case in controls['archive_error_controls']:
            with self.subTest(defect=case['defect']), tempfile.TemporaryDirectory() as scratch:
                directory = Path(scratch)
                packages = {family: directory / family for family in ('debian', 'fedora')}
                for package in packages.values():
                    package.write_bytes(b'literal package stand-in; no native installation')
                receipt = directory / 'deb-receipt.json'
                receipt.write_text(json.dumps({'source': {'original_head': controls['original_head'],
                                                        'source_date_epoch': controls['source_date_epoch']}}))
                args = SimpleNamespace(deb_receipt=receipt, rpm_receipt=receipt, out=directory / 'results/out',
                                       frozen=controls['original_head'], source_date_epoch=controls['source_date_epoch'],
                                       harness_commit='1' * 40)
                def command_output(command):
                    if command == ['dpkg', '--compare-versions', '0.1.0~alpha.0', 'lt', '0.1.0~alpha.1']:
                        return b''
                    if command[0] == 'dpkg-deb':
                        return b'0.1.0~alpha.1'
                    if '--eval' in command:
                        return b'-1'
                    return b'0' if '%{EPOCHNUM}' in command else b'0.1.0-0.alpha.1'
                def corrupt_archive(command, env, out, label):
                    work = Path(command[-1]).parent
                    raw = bytes.fromhex(case['bytes_hex'])
                    (work / 'literal-corrupt-archive.bin').write_bytes(raw)
                    if case['defect'] == 'truncated-control-tar':
                        with tarfile.open(fileobj=io.BytesIO(raw)):
                            pass
                    elif case['defect'] == 'invalid-xz':
                        lzma.decompress(raw)
                    elif case['defect'] == 'missing-member':
                        {}['literal-required-control-member']
                    else:
                        raise EOFError('literal truncated archive member')
                    self.fail('literal archive defect did not raise')
                with patch.dict('sys.modules', {'inputs': SimpleNamespace(validate_inputs=lambda *args: packages)}), \
                     patch.dict(os.environ, {'FABRIC_SCRATCH_ROOT': scratch}), \
                     patch.object(build_candidate_rpm, 'STORAGE', directory), \
                     patch.object(build_candidate_rpm, 'validate_harness'), \
                     patch.object(build_candidate_rpm, 'inventory', return_value={'usr/bin/fabric-server': {'bytes': 3}}), \
                     patch.object(build_candidate_rpm, 'deb_controls', return_value={'preinst': {'bytes': 3}}), \
                     patch.object(build_candidate_rpm, 'rpm_records', return_value={'scripts': {'PREIN': {'bytes': 3}}, 'metadata': {}}), \
                     patch.object(build_candidate_rpm, 'bounded_command', side_effect=command_output), \
                     patch.object(build_candidate_rpm, 'run_build', side_effect=corrupt_archive):
                    escaped = None
                    result = None
                    try:
                        result = build_candidate_rpm.fixture_main(args)
                    except Exception as error:
                        escaped = error
                        self.assertEqual(type(error).__name__, case['exception'])
                report = json.loads((args.out / 'fixture-receipt.json').read_text())
                self.assertIsNone(escaped, 'archive exception escaped without an archived failed verdict')
                self.assertEqual(result, 2)
                self.assertEqual((report['state'], report['exit']), ('failed', 2))
                self.assertTrue(report.get('error'), 'archive failure diagnosis is missing')
                retained = Path(report['failure_evidence'])
                self.assertTrue(retained.is_relative_to(args.out))
                self.assertEqual((retained / 'literal-corrupt-archive.bin').read_bytes(), bytes.fromhex(case['bytes_hex']))
                self.assertEqual(list(directory.glob('owner-fixture-*')), [])
                self.assertEqual(report['packages'], {})

    def test_same_code_fixture_deb_tar_header_controls(self):
        # Origin: #59 C-041 SF4, equal bytes did not bind tar member metadata.
        controls = json.loads((ROOT / 'docs/experiments/formal/fixtures/owner-walkthrough.json').read_text())
        self.assertTrue(callable(getattr(build_candidate_rpm, 'deb_headers', None)),
                        'missing complete fixture Debian tar-header collector')
        original = controls['deb_header_fixture']
        def collect(records):
            archives = {}
            for name, rows in records.items():
                raw = io.BytesIO()
                with tarfile.open(fileobj=raw, mode='w') as archive:
                    for row in rows:
                        member = tarfile.TarInfo(row['name'])
                        member.type = row['type'].encode()
                        for field in ('mode', 'uid', 'gid', 'uname', 'gname', 'mtime', 'size', 'linkname'):
                            setattr(member, field, row[field])
                        archive.addfile(member, io.BytesIO(b'x' * row['size']))
                archives[name] = raw.getvalue()
            with patch.object(build_candidate_rpm, 'bounded_command', side_effect=lambda command: archives[command[-1]]):
                return build_candidate_rpm.deb_headers(Path('literal.deb'))
        self.assertEqual(collect(original), original)
        for archive in original:
            for member in range(len(original[archive])):
                for case in controls['deb_header_controls']:
                    if original[archive][member]['type'] == '5' and case['field'] == 'size':
                        continue
                    with self.subTest(archive=archive, member=member, field=case['field']):
                        changed = json.loads(json.dumps(original))
                        if case['field'] == 'omitted':
                            changed[archive].pop(member)
                        else:
                            changed[archive][member][case['field']] = case['value']
                        self.assertNotEqual(collect(changed), original, 'tar-header drift was invisible')

    def test_same_code_fixture_rpm_provide_and_weak_dependency_controls(self):
        # Origin: #59 C-041 SF3; only the two native self-provide EVRs may change.
        controls = json.loads((ROOT / 'docs/experiments/formal/fixtures/owner-walkthrough.json').read_text())
        def collect(records):
            xml = '<rpmHeader>'
            for name, value in records.items():
                kind, values = value['kind'], value['values']
                item = 'integer' if kind.startswith('integer') else 'string'
                body = ''.join('<' + item + '>' + str(v) + '</' + item + '>' for v in values)
                if kind.endswith('Array'):
                    body = '<' + kind + '>' + body + '</' + kind + '>'
                xml += '<rpmTag name="' + name + '">' + body + '</rpmTag>'
            xml += '</rpmHeader>'
            with patch.object(build_candidate_rpm, 'bounded_command', return_value=xml.encode()):
                return build_candidate_rpm.rpm_records(Path('literal.rpm'))['metadata']
        original = controls['rpm_metadata_fixture']
        expected = collect(original)
        for case in controls['rpm_metadata_controls']:
            with self.subTest(defect=case['defect']):
                changed = json.loads(json.dumps(original))
                changed['RELEASE']['values'][0] = '0.alpha.0'
                changed['PROVIDEVERSION']['values'][:2] = ['0.1.0-0.alpha.0'] * 2
                if case['tag'] is not None:
                    changed[case['tag']]['values'][case['index']] = case['value']
                actual = collect(changed)
                if case['expected'] == 'accepted':
                    self.assertEqual(actual, expected, 'expected self-provide EVR change was refused')
                else:
                    self.assertNotEqual(actual, expected, 'unrelated native metadata drift was invisible')

    def test_same_code_fixture_source_and_literal_transform_controls(self):
        # Origin: owner walkthrough requires a real native version increase
        # without rewriting exact candidate payload, scripts or provenance.
        self.assertTrue(callable(getattr(build_candidate_rpm, 'validate_fixture_source', None)),
                        'missing fixture source validator')
        self.assertTrue(callable(getattr(build_candidate_rpm, 'validate_fixture_transform', None)),
                        'missing fixture transform validator')
        validate_fixture_source = build_candidate_rpm.validate_fixture_source
        validate_fixture_transform = build_candidate_rpm.validate_fixture_transform
        controls = json.loads((ROOT / 'docs/experiments/formal/fixtures/owner-walkthrough.json').read_text())
        source = {'original_head': controls['original_head'], 'source_date_epoch': controls['source_date_epoch']}
        validate_fixture_source(source, controls['original_head'], controls['source_date_epoch'])
        for field, bad in (('original_head', '0' * 40), ('source_date_epoch', 1), ('source_date_epoch', True)):
            with self.subTest(field=field, bad=bad), self.assertRaisesRegex(ValueError, 'fixture source pins differ'):
                validate_fixture_source({**source, field: bad}, controls['original_head'], controls['source_date_epoch'])
        for family, versions in controls['fixture_versions'].items():
            for case in controls['transform_controls']:
                with self.subTest(family=family, defect=case['defect']):
                    before = {'usr/bin/fabric-server': {'sha256': 'a' * 64, 'bytes': 3},
                              'usr/share/fabrico11y/console/index.html': {'sha256': 'b' * 64, 'bytes': 4},
                              'usr/share/doc/fabrico11y/CANDIDATE-PROVENANCE.json': {'sha256': 'c' * 64, 'bytes': 5}}
                    after = {name: dict(value) for name, value in before.items()}
                    scripts, changed = b'exact maintainer scripts', b'exact maintainer scripts'
                    version = versions[1]
                    if case['defect'] == 'payload':
                        after['usr/share/doc/fabrico11y/CANDIDATE-PROVENANCE.json']['sha256'] = 'd' * 64
                    elif case['defect'] == 'script':
                        changed = b'exit 0'
                    elif case['defect'] == 'equal-version':
                        version = versions[0]
                    if case['expected'] == 'accepted':
                        validate_fixture_transform(family, versions[0], version, before, after, scripts, changed)
                    else:
                        with self.assertRaisesRegex(ValueError, case['expected']):
                            validate_fixture_transform(family, versions[0], version, before, after, scripts, changed)

    def test_mpl_source_archive_identity_and_tamper_refusal(self):
        # Origin: release source obligation. A declared version alone cannot
        # establish the identity of the source shipped beside binary code.
        with tempfile.TemporaryDirectory() as scratch:
            directory = Path(scratch)
            archive = directory / 'cargo/registry/cache/registry/webauthn-rs-0.5.5.crate'
            archive.parent.mkdir(parents=True)
            archive.write_bytes(b'exact upstream source fixture')
            expected = hashlib.sha256(archive.read_bytes()).hexdigest()
            lock = directory / 'Cargo.lock'
            lock.write_text([[package]]\nname="webauthn-rs"\nversion="0.5.5"\nsource="registry+fixture"\nchecksum="' + expected + '"\n')
            package = {'name': 'webauthn-rs', 'version': '0.5.5', 'source': 'registry+fixture', 'id': 'fixture'}
            with patch.object(dependency_notices, 'ROOT', directory), patch.dict(os.environ, {'CARGO_HOME': str(directory / 'cargo')}):
                identity = dependency_notices.bundle_mpl_source(package, directory / 'docs')
                self.assertEqual(identity['sha256'], expected)
                self.assertEqual((directory / 'docs' / identity['path']).read_bytes(), archive.read_bytes())
                archive.write_bytes(b'tampered')
                with self.assertRaises(ValueError):
                    dependency_notices.bundle_mpl_source(package, directory / 'bad')
                self.assertFalse((directory / 'bad').exists())
                package['version'] = '0.5.6'
                with self.assertRaises(ValueError):
                    dependency_notices.bundle_mpl_source(package, directory / 'future')

    def test_candidate_rejects_wrong_version_and_substituted_package(self):
        for version in ('0.1.0~alpha.0', '0.1.0~alpha.2', '0.1.0'):
            with self.assertRaises(ValueError):
                validate_control('Package: fabrico11y\nVersion: ' + version + '\nArchitecture: amd64\n')
        with tempfile.TemporaryDirectory() as scratch:
            directory = Path(scratch)
            package = directory / 'candidate.deb'
            package.write_bytes(b'substituted payload')
            receipt = directory / 'receipt.json'
            receipt.write_text(json.dumps({'state': 'built', 'package_family': 'debian', 'container_cleanup_confirmed': True, 'package': str(package), 'package_sha256': '0' * 64}))
            with self.assertRaises(ValueError):
                stage_candidate(package, receipt, directory / 'stage')
            self.assertFalse((directory / 'stage').exists())

    def fixture(self, directory):
        build = directory / 'build'
        dist = build / 'dist'
        dist.mkdir(parents=True)
        assets = {}
        for name in ('index.html', 'service-worker.js', 'manifest.webmanifest', 'icon-192.png', 'icon-512.png', 'client-a.wasm'):
            body = png() if name.endswith('.png') else name.encode()
            (dist / name).write_bytes(body)
            assets[name] = {'bytes': len(body), 'sha256': hashlib.sha256(body).hexdigest()}
        (build / 'build.json').write_text(json.dumps({'exit': 0, 'assets': assets, 'source_sha256': {'crates/fabric-ui/index.html': hashlib.sha256((ROOT / 'crates/fabric-ui/index.html').read_bytes()).hexdigest()}, 'headers': {'Content-Security-Policy': "default-src 'self'"}}))
        return build

    def test_complete_assets_preserve_identity(self):
        with tempfile.TemporaryDirectory() as scratch:
            directory = Path(scratch)
            build = self.fixture(directory)
            stage(build, directory / 'package')
            manifest = json.loads((directory / 'package/asset-manifest.json').read_text())
            for name, identity in manifest.items():
                self.assertEqual(hashlib.sha256((directory / 'package' / name).read_bytes()).hexdigest(), identity['sha256'])
            self.assertTrue((directory / 'package/console-headers.json').is_file())

    def test_tampered_or_missing_or_unlisted_assets_refused(self):
        # Origin: packaging review. Contract: a candidate must ship the exact
        # complete hashed app shell. Defects are deliberately injected.
        for defect in ('changed', 'missing', 'unlisted', 'symlink', 'failed', 'source'):
            with self.subTest(defect=defect), tempfile.TemporaryDirectory() as scratch:
                directory = Path(scratch)
                build = self.fixture(directory)
                if defect == 'changed':
                    (build / 'dist/client-a.wasm').write_bytes(b'corrupt')
                elif defect == 'missing':
                    (build / 'dist/icon-192.png').unlink()
                elif defect == 'unlisted':
                    (build / 'dist/extra.js').write_text('extra')
                elif defect == 'symlink':
                    (build / 'dist/client-a.wasm').unlink()
                    (build / 'dist/client-a.wasm').symlink_to(build / 'dist/index.html')
                elif defect == 'source':
                    receipt = json.loads((build / 'build.json').read_text())
                    receipt['source_sha256']['crates/fabric-ui/index.html'] = '0' * 64
                    (build / 'build.json').write_text(json.dumps(receipt))
                else:
                    receipt = json.loads((build / 'build.json').read_text())
                    receipt['exit'] = 1
                    (build / 'build.json').write_text(json.dumps(receipt))
                with self.assertRaises(ValueError):
                    stage(build, directory / 'package')
                self.assertFalse((directory / 'package').exists())

    def test_console_icons_carry_no_build_timestamp(self):
        # Origin: #55. ImageMagick wrote date:modify/date:timestamp text chunks,
        # so two console builds of one commit shipped different icons and
        # packages. Contract: separate builds of the same source are identical.
        for chunk in (b'tIME', b'tEXt', b'zTXt', b'iTXt'):
            with self.subTest(chunk=chunk), tempfile.TemporaryDirectory() as scratch:
                directory = Path(scratch)
                build = self.fixture(directory)
                body = png(chunk)
                (build / 'dist/icon-192.png').write_bytes(body)
                receipt = json.loads((build / 'build.json').read_text())
                receipt['assets']['icon-192.png'] = {'bytes': len(body), 'sha256': hashlib.sha256(body).hexdigest()}
                (build / 'build.json').write_text(json.dumps(receipt))
                with self.assertRaisesRegex(ValueError, 'time or text metadata'):
                    stage(build, directory / 'package')
                self.assertFalse((directory / 'package').exists())

    def test_preinstall_refuses_root_ids_without_touching_host_accounts(self):
        # Origin: packaging review found the old >=1000 guard admitted UID/GID0.
        # Independent contract: installed service identity must be non-root.
        for uid, gid, expected in ((0, 500, 1), (500, 0, 1), (500, 500, 0)):
            with self.subTest(uid=uid, gid=gid), tempfile.TemporaryDirectory() as scratch:
                directory = Path(scratch)
                getent = directory / 'getent'
                getent.write_text(f"#!/bin/sh\ncase \"$1\" in passwd) echo 'fabricolly:x:{uid}:{gid}:service:/:/usr/sbin/nologin';; group) echo 'fabricolly:x:{gid}:';; esac\n")
                getent.chmod(0o755)
                identity = directory / 'id'
                identity.write_text('#!/bin/sh\necho fabricolly\n')
                identity.chmod(0o755)
                result = subprocess.run(['sh', str(ROOT / 'packaging/debian/preinst'), 'install'],
                                        env=dict(os.environ, PATH=str(directory) + ':' + os.environ['PATH']),
                                        text=True, capture_output=True)
                self.assertEqual(result.returncode, expected)
                if expected:
                    self.assertIn('refusing to install', result.stderr)

    def test_unregistered_predecessor_is_refused_before_extracting(self):
        with tempfile.TemporaryDirectory() as scratch:
            directory = Path(scratch)
            package = directory / 'wrong.deb'
            package.write_bytes(b'unregistered predecessor')
            with self.assertRaises(ValueError):
                stage_predecessor(package, directory / 'stage')
            self.assertFalse((directory / 'stage').exists())

    def test_interrupted_predecessor_reversion_exits_nonzero(self):
        # Origin: review of the predecessor script. Its cleanup trap also caught
        # TERM and then resumed, so a run signalled while packaging hashed the
        # output and exited 0. Stubs stand in for the registered package.
        with tempfile.TemporaryDirectory() as scratch:
            directory = Path(scratch)
            (directory / 'out').mkdir()
            for name, body in (
                ('sha256sum', 'echo "ac71995b3c8b88cbb6f7dd05d9854e347b7a783caadc1b602f6590d6460f4a2f  $1"'),
                ('dpkg-deb', 'if [ "$1" = -R ]; then mkdir -p "$3/DEBIAN"; '
                             'printf "Version: 0\\nArchitecture: amd64\\n" > "$3/DEBIAN/control"; '
                             'else kill -TERM "$PPID"; : > "$5"; fi'),
            ):
                (directory / name).write_text('#!/bin/sh\n' + body + '\n')
                (directory / name).chmod(0o755)
            result = subprocess.run(['sh', str(ROOT / 'packaging/reversion-predecessor-deb.sh'),
                                     str(directory / 'sha256sum'), str(directory / 'out')],
                                    env=dict(os.environ, TMPDIR=scratch,
                                             PATH=scratch + ':' + os.environ['PATH']),
                                    capture_output=True, text=True)
            self.assertEqual(result.returncode, 143, result.stdout + result.stderr)
            self.assertEqual(list((directory / 'out').glob('*.sha256')), [])
            self.assertEqual(list(directory.glob('predecessor-*')), [])

    def test_packagers_have_valid_shell_syntax(self):
        for script in (ROOT / 'packaging').rglob('*'):
            if script.is_file() and script.read_bytes().startswith(b'#!/bin/sh'):
                subprocess.run(['sh', '-n', str(script)], check=True)
                # A cleanup trap that also catches a signal resumes the script.
                self.assertNotRegex(script.read_text(), r'trap .*EXIT.*(HUP|INT|TERM)', script)
        subprocess.run(['bash', '-n', str(ROOT / 'tools/ci/prepare-resource-host.sh')], check=True)

    def test_server_and_companion_bound_is_below_four_decimal_gb(self):
        unit = (ROOT / 'packaging/systemd/fabrico11y-server.service').read_text()
        memory = next(line.split('=', 1)[1] for line in unit.splitlines() if line.startswith('MemoryMax='))
        self.assertTrue(memory.endswith('M'))
        self.assertLessEqual(int(memory[:-1]) * 1024**2, 4_000_000_000)
        self.assertIn('MemorySwapMax=0', unit)
        self.assertIn('Slice=system-fabrico11y.slice', unit)


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

"""Prospective stopped-custody controls; fixture bytes never come from Rust."""
import copy
import hashlib
import importlib.util
import os
from pathlib import Path
import stat
import struct
import tempfile
import unittest
import zlib

ADAPTER = Path(__file__).with_name('custody_witness.py')
SOURCE = '/var/log/fabric-accept/allowed.log'
NODE = bytes(range(1, 17))
PENDING = b'accept-upgrade-pending-literal\n'
UNREAD = b'accept-upgrade-unread-literal\n'
BINDING = dict(run_id='custody-literal',
    candidate_frozen='8fb6f413179b094288316e25845c137440ba23fa',
    candidate_package_sha256='8edfc7709f4b5dc3da01c00a1d8c442470298e96f8c8e502cfb14e8c9e713e92',
    package_source_commit='73ff77867d10ef521ca985b094e1d177f7786b77',
    harness_commit='fcd57f62581a243fad34c61a258acb760cbdfce8')
PROCESS = dict(unit='fabrico11y-node.service', pid=55, boot_id='literal-boot-a',
               start_ns=100, end_ns=200, **BINDING)


def varint(value):
    out = bytearray()
    while value >= 128:
        out.append((value & 127) | 128); value >>= 7
    return bytes(out + bytes([value]))


def vi(tag, value):
    return varint(tag << 3) + varint(value)


def ld(tag, value):
    return varint(tag << 3 | 2) + varint(len(value)) + value


def cursor(offset, *, prefix_crc=None, skipping=0, identity=b''):
    prefix = (PENDING + UNREAD)[:min(offset, 64)]
    return (ld(1, SOURCE.encode()) + vi(2, 7) + vi(3, 9) + vi(4, offset)
            + vi(5, skipping) + vi(6, len(prefix))
            + vi(7, zlib.crc32(prefix) if prefix_crc is None else prefix_crc)
            + (ld(8, identity) if identity else b''))


def batch(sequence=4, *, version=1, node=NODE, generation=1, body=PENDING,
          offset=None, cursor_bytes=None, logs_bytes=None):
    offset = len(PENDING) if offset is None else offset
    attrs = {'log.file.path': SOURCE, 'log.file.device': '7', 'log.file.inode': '9',
             'log.file.offset.start': str(offset - len(body)),
             'log.file.offset.end': str(offset)}
    record = b'\x59' + struct.pack('<Q', 123) + ld(5, ld(1, body[:-1]))
    for key, value in attrs.items():
        record += ld(6, ld(1, key.encode()) + ld(2, ld(1, value.encode())))
    logs = ld(1, ld(2, ld(2, record))) if logs_bytes is None else logs_bytes
    return (vi(1, version) + ld(2, node) + vi(3, generation) + vi(4, sequence)
            + ld(6, logs) + ld(7, cursor(offset) if cursor_bytes is None else cursor_bytes))


def frame(payload, at=0):
    first = b'FAB1' + struct.pack('<I', len(payload))
    marker = b'FAC1' + struct.pack('<Q', at + 16 + len(payload))
    return (first + struct.pack('<II', zlib.crc32(first), zlib.crc32(payload))
            + payload + marker + struct.pack('<I', zlib.crc32(marker)))


def adapter():
    spec = importlib.util.spec_from_file_location('custody_witness', ADAPTER)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


class MechanismAvailability(unittest.TestCase):
    def test_stopped_snapshot_and_conservation_are_available(self):
        self.assertTrue(ADAPTER.is_file(), 'NOT RUN: independent custody adapter is absent')
        module = adapter()
        self.assertTrue(callable(getattr(module, 'stopped_snapshot', None)))
        self.assertTrue(callable(getattr(module, 'conserve', None)))


class CustodyWitnessControls(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        if not ADAPTER.is_file():
            raise unittest.SkipTest('NOT RUN: independent custody adapter is absent')
        cls.module = adapter()

    def setUp(self):
        self.temp = tempfile.TemporaryDirectory(dir=os.environ['FABRIC_SCRATCH_ROOT'])
        self.addCleanup(self.temp.cleanup)
        self.root = Path(self.temp.name)
        archive = self.root / 'allowed.log'; archive.write_bytes(PENDING + UNREAD)
        self.logs = {SOURCE: dict(archive_path=str(archive), device=7, inode=9,
                                 btrfs_identity=None)}

    def snapshot(self, name, payloads=None, *, ack=3, change=None, receipt_change=None):
        root = self.root / name; root.mkdir()
        payloads = [batch()] if payloads is None else payloads
        raw = b''
        for payload in payloads: raw += frame(payload, len(raw))
        (root / 'identity').write_bytes(b'FAI1' + NODE + struct.pack('<Q', 1))
        (root / 'acked').write_bytes(b'FAK1' + struct.pack('<QQ', 1, ack))
        (root / 'writer.lock').touch()
        (root / 'batches.faj').write_bytes(raw)
        receipt = dict(binding=dict(BINDING), spool_root=str(root), spool_bytes=268435456,
            stop_confirmed=True,
            boot_id='literal-boot-a', extra_project_units=[],
            units={unit: dict(active_state='inactive', query_exit=3, main_pid=0,
                cgroup_empty=True, companions=[])
                for unit in ('fabrico11y-server.service', 'fabrico11y-node.service')},
            candidate_process=copy.deepcopy(PROCESS) if name == 'drain' else None)
        if change: change(root)
        if receipt_change: receipt_change(receipt)
        def inventory():
            return {path.name: (path.lstat().st_mode,
                    str(path.readlink()) if path.is_symlink() else
                    path.read_bytes() if stat.S_ISREG(path.lstat().st_mode) else None)
                    for path in root.iterdir()}
        original = inventory()
        try:
            return self.module.stopped_snapshot(root, receipt, self.logs)
        finally:
            self.assertEqual(inventory(), original, 'snapshot reader mutated raw source')

    def chain(self):
        before = self.snapshot('before')
        installed = self.snapshot('installed')
        drained = self.snapshot('drain', [batch(), batch(5, body=UNREAD,
                                offset=len(PENDING + UNREAD))], ack=5)
        old = [dict(node='spindle-1', node_id=NODE.hex(), sequence=seq, index=0,
                    observed_ns=120, body=body, attributes={}) for seq, body in enumerate(
                    ('accept-line-before-start', 'accept-line-during-outage',
                     'accept-line-after-pressure'), 1)]
        rows = copy.deepcopy(old)
        for sequence, body, start in ((4, PENDING, 0), (5, UNREAD, len(PENDING))):
            rows.append(dict(node='spindle-1', node_id=NODE.hex(), sequence=sequence,
                index=0, observed_ns=123, body=body[:-1].decode(), attributes={
                    'log.file.path': SOURCE, 'log.file.device': '7', 'log.file.inode': '9',
                    'log.file.offset.start': str(start), 'log.file.offset.end': str(start + len(body))}))
        queries = dict(before=old, after_drain=copy.deepcopy(rows),
            after_reboot=copy.deepcopy(rows), boot_before='literal-boot-a',
            boot_after='literal-boot-b', pending_body=PENDING[:-1].decode(),
            unread_body=UNREAD[:-1].decode())
        events = [dict(unit=PROCESS['unit'], pid=55, boot_id=PROCESS['boot_id'], at_ns=150,
            message=f'delivery sequence={seq} sha256={hashlib.sha256(payload).hexdigest()} '
                    f'status=ack committed_through={seq} elapsed_us=1')
            for seq, payload in ((4, batch()), (5, batch(5, body=UNREAD,
                                                offset=len(PENDING + UNREAD))))]
        return before, installed, drained, queries, events

    def test_clean_raw_snapshot_and_two_line_conservation(self):
        chain = self.chain()
        self.assertEqual(chain[0]['node_id'], NODE.hex())
        self.assertEqual(chain[0]['ack'], 3)
        self.assertEqual(chain[0]['sources'][0]['sha256'], hashlib.sha256(batch()).hexdigest())
        self.assertEqual(chain[0]['cursors'][SOURCE]['offset'], len(PENDING))
        self.module.conserve(*chain)

    def test_complete_frame_and_envelope_refusals(self):
        wrong_wire = ld(1, b'\x01') + batch()[2:]
        variants = [batch(version=2), wrong_wire, batch(node=NODE[:-1]),
                    batch(generation=0), batch(sequence=0), batch(sequence=2**64 - 1),
                    batch(generation=2**64), batch(cursor_bytes=cursor(len(PENDING), skipping=2))]
        for index, payload in enumerate(variants):
            with self.subTest(payload=index), self.assertRaises(ValueError):
                self.snapshot(f'envelope-{index}', [payload])
        changes = [lambda root: (root / 'batches.faj').write_bytes(frame(batch())[:-1]),
                   lambda root: (root / 'batches.faj').write_bytes(frame(batch()) + b'F'),
                   lambda root: (root / 'batches.faj').write_bytes(frame(batch())[:16] + b'X' + frame(batch())[17:]),
                   lambda root: (root / 'recovery-required').write_bytes(b'failure'),
                   lambda root: (root / 'append-in-progress').write_bytes(b'interrupted'),
                   lambda root: (root / 'acked').write_bytes(b'FAK1' + struct.pack('<QQ', 1, 2**64 - 1)),
                   lambda root: (root / 'acked').write_bytes(b'FAK1' + struct.pack('<QQ', 2, 0))]
        for index, change in enumerate(changes):
            with self.subTest(frame=index), self.assertRaises(ValueError):
                self.snapshot(f'frame-{index}', change=change)

    def test_nested_otlp_structural_refusals(self):
        # Prost permits a sparse record with a valid unknown field, not malformed wire keys.
        positive = self.snapshot('otlp-unknown-valid',
            [batch(logs_bytes=ld(1, ld(2, ld(2, vi(99, 0)))))])
        self.assertEqual(positive['sources'][0]['rows'][0]['body'], '')
        variants = (
            ('field-zero', b'\x00\x00', 'protobuf (field|tag)'),
            ('field-too-large', ld(1, ld(2, ld(2, varint((1 << 29) << 3) + b'\x00'))),
             'protobuf (field|tag)'),
            ('varint-too-long', ld(1, ld(2, ld(2, varint(99 << 3) + b'\x80' * 10 + b'\x00'))),
             'varint'),
        )
        for name, logs, reason in variants:
            with self.subTest(defect=name), self.assertRaisesRegex(ValueError, reason):
                self.snapshot('otlp-' + name, [batch(logs_bytes=logs)])

    def test_full_cursor_prefix_identity_and_sequence_order(self):
        older = (vi(1, 1) + ld(2, NODE) + vi(3, 1) + vi(4, 4)
                 + ld(7, cursor(len(PENDING + UNREAD))) + ld(8, b'older cursor fixture'))
        snapshot = self.snapshot('latest', [older, batch(5)])
        self.assertEqual(snapshot['cursors'][SOURCE]['offset'], len(PENDING))
        self.assertEqual(snapshot['cursors'][SOURCE]['prefix_crc'], zlib.crc32(PENDING))
        for index, raw in enumerate((cursor(len(PENDING), prefix_crc=1),
                                     cursor(len(PENDING), skipping=1),
                                     cursor(len(PENDING), identity=ld(1, b'x') + vi(2, 1)))):
            with self.subTest(cursor=index), self.assertRaises(ValueError):
                self.snapshot(f'cursor-{index}', [batch(cursor_bytes=raw)])
        self.logs[SOURCE]['btrfs_identity'] = dict(uuid=NODE.hex(), subvolume_id=7)
        full = ld(1, NODE) + vi(2, 7)
        valid = self.snapshot('btrfs', [batch(cursor_bytes=cursor(len(PENDING), identity=full))])
        self.assertEqual(valid['cursors'][SOURCE]['btrfs_identity'], self.logs[SOURCE]['btrfs_identity'])

    def test_stop_and_path_admission_refusals(self):
        mutations = [lambda receipt: receipt.update(stop_confirmed=False),
                     lambda receipt: receipt.update(extra_project_units=['unexpected.service']),
                     lambda receipt: receipt['units'].pop('fabrico11y-node.service'),
                     lambda receipt: receipt['units']['fabrico11y-node.service'].update(main_pid=55),
                     lambda receipt: receipt['units']['fabrico11y-node.service'].update(query_exit=1)]
        for index, mutation in enumerate(mutations):
            with self.subTest(stop=index), self.assertRaises(ValueError):
                self.snapshot(f'stop-{index}', receipt_change=mutation)
        target = self.root / 'outside-identity'; target.write_bytes(b'FAI1' + NODE + struct.pack('<Q', 1))
        def symlink(root):
            (root / 'identity').unlink(); (root / 'identity').symlink_to(target)
        with self.assertRaises(ValueError): self.snapshot('symlink', change=symlink)
        self.assertEqual(target.read_bytes(), b'FAI1' + NODE + struct.pack('<Q', 1))

    def test_install_ack_query_and_reboot_conservation_refusals(self):
        chain = self.chain()
        mutations = [lambda args: args[0].update(ack=4),
                     lambda args: args[1]['binding'].update(run_id='other-run'),
                     lambda args: args[1]['binding'].update(candidate_package_sha256='00' * 32),
                     lambda args: args[1].update(node_id='00' * 16),
                     lambda args: args[1]['sources'].clear(),
                     lambda args: args[1]['sources'][0].update(sha256='00' * 32),
                     lambda args: args[1]['cursors'][SOURCE].update(offset=len(PENDING + UNREAD)),
                     lambda args: args[2].update(ack=0),
                     lambda args: args[4].clear(),
                     lambda args: args[4][0].update(pid=44),
                     lambda args: args[4][0].update(unit='other.service'),
                     lambda args: args[4][0].update(boot_id='predecessor-boot'),
                     lambda args: args[4][0].update(at_ns=99),
                     lambda args: args[4][0].update(message=args[4][0]['message'].replace(hashlib.sha256(batch()).hexdigest(), '00' * 32)),
                     lambda args: args[3]['after_drain'].append(copy.deepcopy(args[3]['after_drain'][0])),
                     lambda args: args[3]['after_reboot'][0].update(observed_ns=124),
                     lambda args: args[3].update(boot_after='literal-boot-a')]
        for index, mutation in enumerate(mutations):
            args = copy.deepcopy(chain); mutation(args)
            with self.subTest(conservation=index), self.assertRaises(ValueError):
                self.module.conserve(*args)


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

"""Read stopped raw custody without repairing files or importing Rust codecs."""
import copy
import hashlib
import json
import os
from pathlib import Path
import re
import stat
import struct
import sys
import zlib

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
import query_oracle as oracle

U64 = 2**64 - 1
MAX_BATCH = 1048576
BINDING = ('run_id', 'candidate_frozen', 'candidate_package_sha256',
           'package_source_commit', 'harness_commit')
UNITS = ('fabrico11y-server.service', 'fabrico11y-node.service')


def require(condition, message):
    if not condition:
        raise ValueError(message)


def fields(raw, types):
    try:
        result = oracle.parse_fields(raw)
        require(all(number > 0 for number in result), 'invalid protobuf field')
        for number, values in result.items():
            for wire, value in values:
                require(number not in types or wire == types[number], 'wrong known-field wire type')
                if wire == 0:
                    require(0 <= value <= U64, 'protobuf integer overflow')
        return result
    except oracle.MalformedInput as error:
        raise ValueError(str(error)) from error


def one(values, number, default=0):
    return values.get(number, [(None, default)])[-1][1]


def cursor(raw):
    f = fields(raw, {1: 2, 2: 0, 3: 0, 4: 0, 5: 0, 6: 0, 7: 0, 8: 2})
    path = one(f, 1, b'').decode('utf-8')
    require(len(path.encode()) <= 4096, 'cursor path cap')
    identity = None
    if 8 in f:
        stable = fields(b''.join(value for _, value in f[8]), {1: 2, 2: 0})
        uuid, subvolume = one(stable, 1, b''), one(stable, 2)
        require(len(uuid) == 16 and any(uuid) and subvolume > 0, 'invalid Btrfs identity')
        identity = dict(uuid=uuid.hex(), subvolume_id=subvolume)
    length, crc = one(f, 6), one(f, 7)
    require(all(value <= 2**32 - 1 for number in (6, 7) for _, value in f.get(number, [])), 'cursor u32 overflow')
    require(length <= 64 and length <= one(f, 4) and crc <= 2**32 - 1, 'invalid cursor prefix')
    return dict(path=path, device=one(f, 2), inode=one(f, 3), offset=one(f, 4),
        skipping_oversize=bool(one(f, 5)), prefix_len=length, prefix_crc=crc, btrfs_identity=identity)


def envelope(raw):
    try:
        f = fields(raw, {1: 0, 2: 2, 3: 0, 4: 0, 5: 2, 6: 2, 7: 2, 8: 2, 9: 2})
        require(all(value <= 2**32 - 1 for _, value in f.get(1, [])), 'Batch version u32 overflow')
        require(one(f, 1) == 1 and len(one(f, 2, b'')) == 16
            and one(f, 3) > 0 and 0 < one(f, 4) < U64, 'invalid Batch identity/version')
        signals = [one(f, number, b'') for number in (5, 6, 9)]
        gaps = [value for _, value in f.get(8, [])]
        require(any(signals) or gaps, 'empty Batch')
        require(sum(map(len, signals)) <= MAX_BATCH and len(f.get(7, [])) <= 16
            and len(gaps) <= 130 and all(len(gap) <= 256 for gap in gaps), 'Batch field cap')
        for gap in gaps:
            gap.decode('utf-8')
        cursors = [cursor(value) for _, value in f.get(7, [])]
        batch = oracle.decode_batch(raw)
        rows = oracle.materialize_record('spindle-1', 0, raw).log_rows
        return batch, cursors, rows
    except (oracle.MalformedInput, UnicodeError, TypeError, struct.error) as error:
        raise ValueError(str(error)) from error


def identity(info):
    return tuple(getattr(info, 'st_' + field) for field in
        ('dev', 'ino', 'mode', 'uid', 'gid', 'nlink', 'size', 'mtime_ns', 'ctime_ns'))


def owned(path):
    path = Path(path).absolute()
    scratch = Path(os.environ['FABRIC_SCRATCH_ROOT']).absolute()
    require(path != scratch and path.is_relative_to(scratch), 'outside owned scratch')
    require(path.resolve(strict=True) == path and scratch.resolve(strict=True) == scratch,
            'symlink or noncanonical input')
    return path


def open_owned(path, flags):
    path = owned(path)
    fd = os.open('/', os.O_RDONLY | os.O_DIRECTORY)
    try:
        for part in path.parts[1:-1]:
            next_fd = os.open(part, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=fd)
            os.close(fd)
            fd = next_fd
        return os.open(path.name, flags | os.O_NOFOLLOW, dir_fd=fd)
    finally:
        os.close(fd)


def read_regular(path, limit):
    before = path.lstat()
    require(stat.S_ISREG(before.st_mode) and before.st_nlink == 1 and before.st_size <= limit,
            'nonregular, aliased or oversized input')
    fd = open_owned(path, os.O_RDONLY)
    try:
        require(identity(os.fstat(fd)) == identity(before), 'input changed before read')
        with os.fdopen(os.dup(fd), 'rb') as stream:
            raw = stream.read(limit + 1)
        require(len(raw) == before.st_size and identity(os.fstat(fd)) == identity(before)
            and identity(path.lstat()) == identity(before), 'input changed during read')
        return raw, identity(before)
    finally:
        os.close(fd)


def stopped_snapshot(root, stopped_receipt, archived_allowed_logs):
    root = owned(root)
    receipt = copy.deepcopy(stopped_receipt)
    require(receipt.get('spool_root') == str(root) and receipt.get('stop_confirmed') is True
        and receipt.get('extra_project_units') == [] and isinstance(receipt.get('boot_id'), str)
        and receipt['boot_id'], 'incomplete stop receipt')
    binding = receipt.get('binding', {})
    require(all(isinstance(binding.get(key), str) and binding[key] for key in BINDING), 'missing binding')
    for key in BINDING[1:]:
        require(re.fullmatch('[0-9a-f]{64}' if key.endswith('sha256') else '[0-9a-f]{40}',
                binding[key]), 'invalid source/package binding')
    require(set(receipt.get('units', {})) == set(UNITS), 'missing or extra unit observation')
    for unit in receipt['units'].values():
        require(unit.get('active_state') in ('inactive', 'failed') and unit.get('query_exit') == 3
            and type(unit.get('query_exit')) is int and unit.get('main_pid') == 0
            and type(unit.get('main_pid')) is int and unit.get('cgroup_empty') is True
            and unit.get('companions') == [], 'unit not confirmed stopped')
    cap = receipt.get('spool_bytes')
    require(type(cap) is int and cap == 268435456, 'unregistered Spool cap')
    root_before = identity(root.lstat())
    require(stat.S_ISDIR(root.lstat().st_mode), 'Spool root is not a directory')
    fd = open_owned(root, os.O_RDONLY | os.O_DIRECTORY)
    try:
        require(identity(os.fstat(fd)) == root_before, 'Spool root changed')
        names = sorted(os.listdir(fd))
        require({'identity', 'writer.lock', 'batches.faj'} <= set(names), 'missing Spool input')
        require(all(name in ('identity', 'acked', 'writer.lock', 'batches.faj')
            or re.fullmatch(r'sealed-[0-9]{20}\.faj', name) for name in names), 'unexpected or failed Spool input')
        observed, files, raw_files = {}, {}, {}
        require(sum((root / name).lstat().st_size for name in names if name.endswith('.faj')) <= cap - 4096,
                'framed Spool cap')
        for name in names:
            raw, info = read_regular(root / name, cap - 4096 if name.endswith('.faj') else 28)
            raw_files[name], observed[name] = raw, info
            files[name] = dict(sha256=hashlib.sha256(raw).hexdigest(), size=len(raw),
                              mode=info[2], uid=info[3], gid=info[4])
        ident = raw_files['identity']
        require(len(ident) == 28 and ident[:4] == b'FAI1', 'invalid identity sidecar')
        node, generation = ident[4:20], struct.unpack('<Q', ident[20:])[0]
        require(generation > 0, 'zero generation')
        ack = 0
        if 'acked' in raw_files:
            acked = raw_files['acked']
            require(len(acked) == 20 and acked[:4] == b'FAK1', 'invalid ACK sidecar')
            ack_generation, ack = struct.unpack('<QQ', acked[4:])
            require(ack_generation == generation and ack < U64, 'ACK identity/exhaustion')
        sources, cursors = [], {}
        ordered = sorted(name for name in names if name.startswith('sealed-')) + ['batches.faj']
        for name in ordered:
            raw, at, first = raw_files[name], 0, True
            while at < len(raw):
                require(len(raw) - at >= 16 and raw[at:at + 4] == b'FAB1', 'incomplete frame/header')
                size, head_crc, payload_crc = struct.unpack('<III', raw[at + 4:at + 16])
                end = at + 16 + size
                require(0 < size <= MAX_BATCH and end + 16 <= len(raw)
                    and zlib.crc32(raw[at:at + 8]) == head_crc, 'frame length/header CRC')
                payload, marker = raw[at + 16:end], raw[end:end + 16]
                require(zlib.crc32(payload) == payload_crc and marker[:4] == b'FAC1'
                    and struct.unpack('<Q', marker[4:12])[0] == end
                    and zlib.crc32(marker[:12]) == struct.unpack('<I', marker[12:])[0], 'payload/commit CRC')
                batch, current, rows = envelope(payload)
                sequence = batch['sequence']
                require(batch['node_id'] == node and batch['generation'] == generation, 'Batch/Spool identity')
                require(not sources or sequence == sources[-1]['sequence'] + 1, 'lost/reordered source')
                require(not first or name == 'batches.faj' or int(name[7:-4]) == sequence, 'sealed label')
                sources.append(dict(sequence=sequence, sha256=hashlib.sha256(payload).hexdigest(),
                                    payload=payload, rows=rows))
                cursors.update((item['path'], item) for item in current)
                at, first = end + 16, False
            require(name == 'batches.faj' or not first, 'empty sealed source')
        require(not sources or sources[0]['sequence'] <= ack + 1 <= sources[-1]['sequence'] + 1,
                'ACK beyond source or missing pending prefix')
        logs, log_observations = {}, {}
        for path, entry in archived_allowed_logs.items():
            require(Path(path).is_absolute(), 'nonabsolute guest source')
            raw, info = read_regular(owned(entry['archive_path']), MAX_BATCH)
            log_observations[entry['archive_path']] = info
            require(all(type(entry[key]) is int and 0 <= entry[key] <= U64 for key in ('device', 'inode')), 'source identity type')
            logs[path] = dict(device=entry['device'], inode=entry['inode'],
                btrfs_identity=copy.deepcopy(entry.get('btrfs_identity')), bytes=raw,
                sha256=hashlib.sha256(raw).hexdigest(), size=len(raw))
        for path, current in cursors.items():
            require(path in logs, 'cursor source not archived')
            log = logs[path]
            namespace = (current['btrfs_identity'] == log['btrfs_identity']
                if current['btrfs_identity'] is not None else current['device'] == log['device'])
            offset = current['offset']
            require(namespace and current['inode'] == log['inode'] and offset <= log['size']
                and not current['skipping_oversize'] and (offset == 0 or log['bytes'][offset - 1:offset] == b'\n')
                and zlib.crc32(log['bytes'][:current['prefix_len']]) == current['prefix_crc'], 'latest cursor/source mismatch')
        require(owned(root) == root and sorted(os.listdir(fd)) == names and identity(os.fstat(fd)) == root_before
            and identity(root.lstat()) == root_before, 'Spool changed during snapshot')
        require(all(identity((root / name).lstat()) == info for name, info in observed.items()), 'Spool file drift')
        require(all(identity(owned(path).lstat()) == info for path, info in log_observations.items()), 'archive drift')
        return dict(binding=binding, boot_id=receipt['boot_id'], node_id=node.hex(), generation=generation,
            ack=ack, files=files, file_observations=observed, sources=sources, cursors=cursors,
            logs=logs, log_observations=log_observations, candidate_process=receipt.get('candidate_process'))
    except (OSError, KeyError, TypeError) as error:
        raise ValueError(str(error)) from error
    finally:
        os.close(fd)


def conserve(before, after_install, after_drain, queries, candidate_ack_events):
    for snapshot in (after_install, after_drain):
        require(all(snapshot[key] == before[key] for key in ('binding', 'node_id', 'generation', 'boot_id')),
                'candidate/Strand/boot substitution')
    require(all(before[key] == after_install[key] for key in ('ack', 'files', 'sources', 'cursors', 'logs')),
            'stopped install changed retained custody')
    pending = [source for source in before['sources'] if source['sequence'] > before['ack']]
    require(pending and after_drain['ack'] >= pending[-1]['sequence'], 'missing pending custody/drain ACK')
    require(before['logs'] == after_drain['logs'], 'source archive rewritten')
    retained = {source['sequence']: source['sha256'] for source in before['sources']}
    require(all(source['sequence'] not in retained or source['sha256'] == retained[source['sequence']]
        for source in after_drain['sources']), 'retained committed bytes changed')
    def sentinel(snapshot, body):
        rows = [row for source in snapshot['sources'] for row in source['rows'] if row['body'] == body]
        require(len(rows) == 1, 'missing/duplicate raw sentinel')
        row, attrs = rows[0], rows[0]['attributes']
        path = attrs.get('log.file.path')
        require(path in snapshot['logs'] and path in snapshot['cursors'], 'sentinel source absent')
        log = snapshot['logs'][path]
        start, end = int(attrs['log.file.offset.start']), int(attrs['log.file.offset.end'])
        require(0 <= start < end <= snapshot['cursors'][path]['offset']
            and (start == 0 or log['bytes'][start - 1:start] == b'\n')
            and log['bytes'][start:end] == (body + '\n').encode()
            and attrs['log.file.device'] == str(log['device']) and attrs['log.file.inode'] == str(log['inode']),
                'sentinel exact source interval mismatch')
        return row, path, start, end
    pending_row, path, _, end = sentinel(before, queries['pending_body'])
    require(pending_row['sequence'] > before['ack'], 'sentinel already acknowledged')
    require(not any(row['body'] == queries['unread_body'] for source in before['sources'] for row in source['rows']),
            'unread sentinel already committed')
    require(before['cursors'][path]['offset'] == end
        and before['logs'][path]['bytes'][end:] == (queries['unread_body'] + '\n').encode(), 'unread suffix mismatch')
    unread_row, unread_path, start, unread_end = sentinel(after_drain, queries['unread_body'])
    require(unread_path == path and start == end and unread_row['sequence'] > before['sources'][-1]['sequence']
        and after_drain['cursors'][path]['offset'] == unread_end == after_drain['logs'][path]['size'], 'unread not drained')
    old = queries['before']
    require(len(old) == 3 and queries['pending_body'] != queries['unread_body']
        and not any(row['body'] in (queries['pending_body'], queries['unread_body']) for row in old), 'invalid old-row baseline')
    expected = old + [pending_row, unread_row]
    require(all(row in old + [pending_row] for source in pending for row in source['rows']), 'unexpected pending log row')
    keys = lambda rows: sorted((row['node_id'], row['sequence'], row['index']) for row in rows)
    encoded = lambda rows: sorted(json.dumps(row, sort_keys=True, allow_nan=False) for row in rows)
    require(len(set(keys(expected))) == len(expected), 'duplicate expected identity')
    for observed in (queries['after_drain'], queries['after_reboot']):
        require(keys(observed) == keys(expected) and encoded(observed) == encoded(expected), 'exact row conservation')
    require(queries['boot_before'] == before['boot_id'] and queries['boot_after']
        and queries['boot_after'] != queries['boot_before'], 'no actual reboot identity')
    process = after_drain['candidate_process']
    require(isinstance(process, dict) and all(process.get(key) == before['binding'][key] for key in BINDING)
        and process.get('unit') == UNITS[1] and type(process.get('pid')) is int and process['pid'] > 0
        and process.get('boot_id') == after_drain['boot_id'] and type(process.get('start_ns')) is int
        and type(process.get('end_ns')) is int and 0 <= process['start_ns'] <= process['end_ns'], 'candidate sender not bound')
    required = {source['sequence']: source['sha256'] for source in pending}
    required[unread_row['sequence']] = next(source['sha256'] for source in after_drain['sources']
                                          if source['sequence'] == unread_row['sequence'])
    witnessed = set()
    for event in candidate_ack_events:
        match = re.fullmatch(r'delivery sequence=([0-9]+) sha256=([0-9a-f]{64}) status=ack committed_through=([0-9]+) elapsed_us=([0-9]+)', event['message'])
        if not match:
            continue
        sequence, digest, through, _ = match.groups()
        sequence, through = int(sequence), int(through)
        require(all(event.get(key) == process[key] for key in ('unit', 'pid', 'boot_id'))
            and type(event.get('at_ns')) is int and process['start_ns'] <= event['at_ns'] <= process['end_ns'], 'ACK sender/window mismatch')
        if sequence in required:
            require(digest == required[sequence] and sequence <= through <= after_drain['ack'], 'ACK exact bytes/cursor mismatch')
            witnessed.add(sequence)
    require(witnessed == set(required), 'missing candidate exact-byte ACK')