from contextlib import contextmanager
import ctypes
import os
from pathlib import Path
import signal
import shutil
import stat
import subprocess
import tempfile
import time
import unittest
import sys
from unittest.mock import patch
import payload
from payload import cpio, identity


def fixture_live(directory):
    pid, start = map(int, (directory / 'child.pid').read_text().split())
    try:
        fields = Path(f'/proc/{pid}/stat').read_text().rsplit(')', 1)[1].split()
    except FileNotFoundError:
        return False
    if int(fields[19]) != start:
        raise AssertionError('owned fixture PID identity changed')
    return fields[0] != 'Z'


@contextmanager
def owned_process_fixture():
    sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
    from resource_group import require_limits
    require_limits()
    libc, previous = ctypes.CDLL(None, use_errno=True), ctypes.c_int()
    if libc.prctl(37, ctypes.byref(previous), 0, 0, 0) or libc.prctl(36, 1, 0, 0, 0):
        raise RuntimeError('owned process control requires Linux subreaper')
    try:
        directory = Path(tempfile.mkdtemp(dir=os.environ['FABRIC_SCRATCH_ROOT'], prefix='package-process-control-'))
        script = directory / 'child.py'
        script.write_text('''import os, select, signal, sys, time
from pathlib import Path
directory, mode = Path(sys.argv[1]), sys.argv[2]
stage = Path(sys.argv[3]) if len(sys.argv) > 3 else directory / 'owned-stage'
expires = time.monotonic() + 45
if os.getpgrp() != os.getpid(): os.setsid()
def atomic(name, body):
    temporary = directory / (name + '.pending')
    with temporary.open('x') as stream:
        stream.write(body); stream.flush(); os.fsync(stream.fileno())
    temporary.replace(directory / name)
def identity(name, pid):
    process = Path(f'/proc/{pid}')
    fields = (process / 'stat').read_text().rsplit(')', 1)[1].split()
    if process.stat().st_uid != os.getuid() or int(fields[19]) <= 0:
        raise RuntimeError('unowned fixture identity')
    if pid != os.getpid() and (int(fields[1]) != os.getpid() or int(fields[2]) != os.getpid()):
        raise RuntimeError('fork child is outside owned leader')
    atomic(name, f'{pid} {fields[19]}')
identity('leader.pid', os.getpid())
if mode == 'drain': signal.signal(signal.SIGTERM, signal.SIG_IGN)
gate_read, gate_write = os.pipe()
child = os.fork()
if child == 0:
    os.close(gate_write)
    if not select.select([gate_read], [], [], 5)[0] or os.read(gate_read, 1) != b'1': os._exit(1)
    os.close(gate_read)
    signal.signal(signal.SIGTERM, signal.SIG_IGN)
    with open(os.devnull, 'wb') as null:
        os.dup2(null.fileno(), 1); os.dup2(null.fileno(), 2)
    stage.mkdir(exist_ok=True)
    held = (stage / 'held-stage.bin').open('wb')
    held.write(b'literal owned staging'); held.flush()
    atomic('ready', 'ready')
    time.sleep(max(0, expires - time.monotonic()))
    os._exit(0)
os.close(gate_read)
try:
    identity('child.pid', child)
    os.write(gate_write, b'1')
finally:
    os.close(gate_write)
deadline = time.monotonic() + 5
while not (directory / 'ready').exists():
    if time.monotonic() >= deadline: raise RuntimeError('child readiness deadline')
    time.sleep(.01)
print('literal process stdout', flush=True)
print('literal process stderr', file=sys.stderr, flush=True)
atomic('diagnostics-ready', 'ready')
if mode != 'success': time.sleep(max(0, expires - time.monotonic()))
''')
        try:
            yield directory, [sys.executable, '-B', str(script), str(directory)]
        finally:
            cleanup_errors = []
            for name in ('leader.pid', 'child.pid'):
                try:
                    marker = directory / name
                    info = marker.lstat()
                    if not stat.S_ISREG(info.st_mode) or info.st_uid != os.getuid() or info.st_nlink != 1 or info.st_size > 128:
                        raise AssertionError('invalid owned fixture identity marker')
                    pid, start = map(int, marker.read_text().split())
                    if pid <= 1 or start <= 0:
                        raise AssertionError('invalid owned fixture PID/start identity')
                    try:
                        descriptor = os.pidfd_open(pid)
                    except ProcessLookupError:
                        continue
                    try:
                        try:
                            fields = Path(f'/proc/{pid}/stat').read_text().rsplit(')', 1)[1].split()
                        except FileNotFoundError:
                            continue
                        if int(fields[19]) != start:
                            raise AssertionError('refuse cleanup of reused fixture PID')
                        try:
                            signal.pidfd_send_signal(descriptor, signal.SIGKILL)
                        except ProcessLookupError:
                            pass
                        deadline = time.monotonic() + 5
                        while True:
                            try:
                                reaped, _ = os.waitpid(pid, os.WNOHANG)
                            except ChildProcessError:
                                reaped = 0
                            if reaped == pid:
                                break
                            try:
                                fields = Path(f'/proc/{pid}/stat').read_text().rsplit(')', 1)[1].split()
                            except FileNotFoundError:
                                break
                            if int(fields[19]) != start:
                                raise AssertionError('fixture PID changed before confirmed reap')
                            if time.monotonic() >= deadline:
                                raise AssertionError('owned fixture cleanup deadline')
                            time.sleep(.01)
                    finally:
                        os.close(descriptor)
                except (OSError, ValueError, IndexError, AssertionError) as error:
                    cleanup_errors.append(name + ': ' + str(error))
            if cleanup_errors:
                raise AssertionError('fixture scratch retained at ' + str(directory) + ': ' + '; '.join(cleanup_errors))
            shutil.rmtree(directory)
    finally:
        if libc.prctl(36, previous.value, 0, 0, 0):
            raise RuntimeError('restore test subreaper setting')


def assert_fixture_stopped(test, directory):
    test.assertFalse(fixture_live(directory), 'command returned with live owned descendant')


def entry(name, body=b'', mode=stat.S_IFREG | 0o644):
    name = name.encode() + b'\0'
    fields = [1, mode, 0, 0, 1, 0, len(body), 0, 0, 0, 0, len(name), 0]
    header = b'070701' + ''.join(f'{value:08x}' for value in fields).encode()
    result = header + name
    result += b'\0' * (-len(result) % 4)
    result += body
    return result + b'\0' * (-len(result) % 4)


class PayloadTests(unittest.TestCase):
    def test_extraction_stops_real_descendant_on_success_and_timeout(self):
        # Origin: static cleanup review; literal child ignores TERM and closes pipes.
        for mode in ('success', 'timeout'):
            with self.subTest(mode=mode), owned_process_fixture() as (directory, command):
                if mode == 'success':
                    self.assertEqual(payload.bounded_command(command + [mode]), b'literal process stdout\n')
                else:
                    with self.assertRaises(subprocess.TimeoutExpired) as caught:
                        payload.bounded_command(command + [mode])
                    self.assertEqual(caught.exception.timeout, 30)
                    self.assertIn(b'literal process stderr', caught.exception.stderr)
                assert_fixture_stopped(self, directory)

    def test_extraction_failure_retains_stderr_and_command(self):
        # Origin: #59 C-041 nit4; the bounded reader discarded failure stderr.
        command = [sys.executable, '-c', 'import sys; sys.stderr.write("literal archive diagnostic\\n"); sys.exit(7)']
        with self.assertRaises(ValueError) as caught:
            payload.bounded_command(command)
        self.assertIn('literal archive diagnostic', str(caught.exception))
        self.assertIn(sys.executable, str(caught.exception))

    def test_extraction_stops_at_cap_before_materializing_oversized_output(self):
        with patch.object(payload, 'CAP', 1024):
            self.assertEqual(payload.bounded_command([sys.executable, '-c', 'print("small")']), b'small\n')
            with self.assertRaisesRegex(ValueError, 'bounded inventory cap'):
                payload.bounded_command([sys.executable, '-c', 'import sys; sys.stdout.buffer.write(b"x"*4096)'])

    def test_exact_content_inventory_with_complete_trailer(self):
        raw = entry('./usr/bin/fabric-node', b'ELF') + entry('TRAILER!!!')
        self.assertEqual(cpio(raw), {'usr/bin/fabric-node': identity(b'ELF')})
        for invalid in [raw[:-100], raw.replace(b'070701', b'070702', 1),
                        entry('../usr/bin/bad', b'bad') + entry('TRAILER!!!'),
                        entry('/usr/bin/bad', b'bad') + entry('TRAILER!!!'),
                        entry('usr/bin/link', b'elsewhere', stat.S_IFLNK | 0o777) + entry('TRAILER!!!'),
                        entry('usr/bin/x', b'a') + entry('usr/bin/x', b'b') + entry('TRAILER!!!')]:
            with self.subTest(invalid=invalid[:15]), self.assertRaises(ValueError):
                cpio(invalid)


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

"""Packaging regressions: exact assets, tamper rejection and bounded unit policy."""
import hashlib
import io
import json
import lzma
import os
from pathlib import Path
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_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()

#!/usr/bin/env python3
"""Record exact candidate RPM construction or lower metadata owner fixtures."""
import argparse
import io
import json
import os
import re
import shutil
import signal
from pathlib import Path
import subprocess
import sys
import tarfile
import tempfile
import xml.etree.ElementTree as ET

HERE = Path(__file__).resolve().parent
ROOT = HERE.parents[1]
sys.path.insert(0, str(HERE.parent))
from resource_group import require_limits, STORAGE
from stage_candidate import digest, stage_candidate
from payload import assert_equivalent, bounded_command, cpio, identity, inventory

FIXTURE_VERSIONS = {'debian': ('0.1.0~alpha.1', '0.1.0~alpha.0'),
                    'fedora': ('0.1.0-0.alpha.1', '0.1.0-0.alpha.0')}


def validate_fixture_source(source, frozen, epoch):
    if (not isinstance(source, dict) or not isinstance(frozen, str)
            or not re.fullmatch('[0-9a-f]{40}', frozen) or type(epoch) is not int or epoch <= 0
            or source.get('original_head') != frozen or type(source.get('source_date_epoch')) is not int
            or source.get('source_date_epoch') != epoch):
        raise ValueError('fixture source pins differ')


def validate_fixture_transform(family, candidate, fixture, before, after, scripts, changed_scripts):
    if (candidate, fixture) != FIXTURE_VERSIONS.get(family):
        raise ValueError('fixture native metadata differs')
    if not before or before != after:
        raise ValueError('fixture payload differs')
    if not scripts or scripts != changed_scripts:
        raise ValueError('fixture scripts differ')


def run_build(command, env, out, label='build'):
    process = subprocess.Popen(command, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                               text=True, start_new_session=True)
    interruption = None
    try:
        stdout, stderr = process.communicate(timeout=300)
    except (subprocess.TimeoutExpired, KeyboardInterrupt) as error:
        interruption = error
        try:
            os.killpg(process.pid, signal.SIGTERM)
        except ProcessLookupError:
            pass
        try:
            stdout, stderr = process.communicate(timeout=10)
        except subprocess.TimeoutExpired:
            try:
                os.killpg(process.pid, signal.SIGKILL)
            except ProcessLookupError:
                pass
            stdout, stderr = process.communicate(timeout=10)
    (out / (label + '.stdout')).write_text(stdout)
    (out / (label + '.stderr')).write_text(stderr)
    if interruption is not None:
        if isinstance(interruption, subprocess.TimeoutExpired):
            interruption.output, interruption.stderr = stdout, stderr
        raise interruption
    return process.returncode


def deb_controls(package):
    result = {}
    size = 0
    raw = bounded_command(['ar', 'p', str(package), 'control.tar.xz'])
    with tarfile.open(fileobj=io.BytesIO(raw)) as archive:
        for member in archive:
            if member.isdir():
                continue
            name = member.name.removeprefix('./')
            size += member.size
            if (not member.isfile() or not name or '/' in name or name in ('.', '..')
                    or name in result or size > 1024**2):
                raise ValueError('unsafe fixture Debian control entry')
            body = archive.extractfile(member).read()
            if name == 'control':
                if sum(line.startswith(b'Version:') for line in body.splitlines()) != 1:
                    raise ValueError('fixture native metadata differs')
                body = b''.join(line for line in body.splitlines(keepends=True) if not line.startswith(b'Version:'))
            result[name] = {'mode': member.mode, **identity(body)}
    return result


def deb_headers(package):
    fields = ('name', 'mode', 'uid', 'gid', 'uname', 'gname', 'mtime', 'size', 'linkname')
    result = {}
    for name in ('control.tar.xz', 'data.tar.xz'):
        raw = bounded_command(['ar', 'p', str(package), name])
        with tarfile.open(fileobj=io.BytesIO(raw)) as archive:
            result[name] = [{**{field: getattr(member, field) for field in fields},
                             'type': member.type.decode('ascii')} for member in archive]
    return result


def rpm_records(package):
    raw = bounded_command(['rpm', '-qp', '--xml', str(package)])
    header = ET.fromstring(raw)
    prefixes = ('PREIN', 'POSTIN', 'PREUN', 'POSTUN', 'PRETRANS', 'POSTTRANS',
                'VERIFYSCRIPT', 'TRIGGER', 'FILETRIGGER', 'TRANSFILETRIGGER')
    blocks = {match[1].decode().upper(): match[0]
              for match in re.finditer(rb'<rpmTag\b[^>]*\bname="([^"]+)"[^>]*>.*?</rpmTag>', raw, re.S)}
    stable = {'NAME', 'VERSION', 'ARCH', 'OS', 'LICENSE', 'SUMMARY', 'DESCRIPTION',
              'FILEMODES', 'FILEFLAGS', 'FILEVERIFYFLAGS', 'FILEUSERNAME', 'FILEGROUPNAME',
              'FILELINKTOS', 'FILECAPS', 'FILEMTIMES', 'BASENAMES', 'DIRNAMES', 'DIRINDEXES',
              'REQUIRENAME', 'REQUIREFLAGS', 'REQUIREVERSION', 'CONFLICTNAME', 'CONFLICTFLAGS',
              'CONFLICTVERSION', 'OBSOLETENAME', 'OBSOLETEFLAGS', 'OBSOLETEVERSION',
              'PROVIDENAME', 'PROVIDEFLAGS', 'PROVIDEVERSION'}
    stable.update(stem + field for stem in ('RECOMMEND', 'SUGGEST', 'SUPPLEMENT', 'ENHANCE')
                  for field in ('NAME', 'FLAGS', 'VERSION'))
    if not {'NAME', 'VERSION', 'ARCH', 'FILEMODES', 'FILEUSERNAME', 'FILEGROUPNAME',
            'PROVIDENAME', 'PROVIDEFLAGS', 'PROVIDEVERSION'} <= blocks.keys():
        raise ValueError('fixture RPM native metadata unavailable')
    nodes = {node.attrib['name'].upper(): node for node in header.findall('rpmTag')}
    names = [node.text for node in nodes['PROVIDENAME'].iter('string')]
    flags = [node.text for node in nodes['PROVIDEFLAGS'].iter('integer')]
    versions = [node.text for node in nodes['PROVIDEVERSION'].iter('string')]
    if not len(names) == len(flags) == len(versions):
        raise ValueError('fixture RPM provide arrays differ')
    provides = iter(zip(names, flags, versions))
    def normalize_self(match):
        name, flag, version = next(provides)
        if (name in ('fabrico11y', 'fabrico11y(x86-64)') and flag == '8'
                and version == '0.1.0-0.alpha.1' and match[1] == b'0.1.0-0.alpha.1'):
            return match[0].replace(b'0.1.0-0.alpha.1', b'0.1.0-0.alpha.0')
        return match[0]
    blocks['PROVIDEVERSION'] = re.sub(rb'<string>(.*?)</string>', normalize_self,
                                     blocks['PROVIDEVERSION'], flags=re.S)
    tags = {name: identity(block) for name, block in blocks.items()}
    return {'scripts': {name: value for name, value in tags.items() if name.startswith(prefixes)},
            'metadata': {name: value for name, value in tags.items() if name in stable}}


def validate_harness(commit):
    if (not re.fullmatch('[0-9a-f]{40}', commit)
            or bounded_command(['git', '-C', str(ROOT), 'rev-parse', 'HEAD']).decode().strip() != commit
            or bounded_command(['git', '-C', str(ROOT), 'status', '--porcelain'])):
        raise ValueError('fixture requires the exact clean pinned harness')


def fixture_main(args):
    if any(value is None for value in (args.rpm_receipt, args.frozen, args.source_date_epoch, args.harness_commit)):
        raise ValueError('fixture requires RPM receipt, frozen head, epoch and full harness commit')
    validate_harness(args.harness_commit)
    source = json.loads(args.deb_receipt.read_text())['source']
    validate_fixture_source(source, args.frozen, args.source_date_epoch)
    sys.path.insert(0, str(HERE.parent / 'qualification/cross_family'))
    from inputs import validate_inputs
    admitted = validate_inputs(args.deb_receipt, args.rpm_receipt)
    packages = {family: admitted[family] for family in ('debian', 'fedora')}
    inputs = {str(path.resolve()): digest(path) for path in (args.deb_receipt, args.rpm_receipt, *packages.values())}
    before = {family: inventory(package) for family, package in packages.items()}
    rpm_before = rpm_records(packages['fedora'])
    scripts = {'debian': deb_controls(packages['debian']), 'fedora': rpm_before['scripts']}
    versions = FIXTURE_VERSIONS
    version_commands = {'debian': ['dpkg-deb', '-f', str(packages['debian']), 'Version'],
                        'fedora': ['rpm', '-qp', '--qf', '%{VERSION}-%{RELEASE}', str(packages['fedora'])]}
    for family, command in version_commands.items():
        if bounded_command(command).decode().strip() != versions[family][0]:
            raise ValueError('fixture native metadata differs')
        validate_fixture_transform(family, *versions[family], before[family], before[family],
                                   scripts[family], scripts[family])
    if bounded_command(['rpm', '-qp', '--qf', '%{EPOCHNUM}', str(packages['fedora'])]).strip() != b'0':
        raise ValueError('fixture RPM epoch differs')
    deb_compare = ['dpkg', '--compare-versions', versions['debian'][1], 'lt', versions['debian'][0]]
    deb_order = bounded_command(deb_compare)
    if deb_order:
        raise ValueError('unexpected Debian version comparison output')
    rpm_compare = ['rpm', '--eval', "%{lua: print(rpm.vercmp('0:0.1.0-0.alpha.0','0:0.1.0-0.alpha.1'))}"]
    rpm_order = bounded_command(rpm_compare)
    if rpm_order.decode().strip() != '-1':
        raise ValueError('fixture RPM version is not natively older')
    out = args.out.resolve()
    out.relative_to(STORAGE / 'results')
    out.mkdir(parents=True, exist_ok=False)
    record = {'state': 'failed', 'exit': 2, 'role': 'same implementation, lower native metadata only; not historical migration',
              'source': source, 'harness_commit': args.harness_commit, 'inputs': inputs,
              'constructor_sha256': digest(Path(__file__)), 'payload_parser_sha256': digest(HERE / 'payload.py'),
              'input_validator_sha256': digest(HERE.parent / 'qualification/cross_family/inputs.py'),
              'rpm_builder_sha256': digest(ROOT / 'packaging/build-rpm.sh'), 'commands': [], 'packages': {},
              'native_comparisons': [{'command': deb_compare, 'exit': 0, 'stdout': deb_order.decode()},
                                     {'command': rpm_compare, 'exit': 0, 'stdout': rpm_order.decode()}]}
    work = None
    env = {**os.environ, 'SOURCE_DATE_EPOCH': str(args.source_date_epoch)}
    try:
        work = Path(tempfile.mkdtemp(dir=os.environ['FABRIC_SCRATCH_ROOT'], prefix='owner-fixture-'))
        deb = out / 'fabrico11y_0.1.0~alpha.0_amd64.deb'
        def execute(command, label):
            step = {'command': command, 'exit': None}
            record['commands'].append(step)
            status = run_build(command, env, out, label)
            step['exit'] = status
            if status:
                if status < 0:
                    record['state'] = 'interrupted'
                raise ValueError('fixture build failed: ' + label)
        execute(['dpkg-deb', '-R', str(packages['debian']), str(work / 'deb')], 'deb-unpack')
        record['debian_tar_headers_before'] = deb_headers(packages['debian'])
        control = work / 'deb/DEBIAN/control'
        body = control.read_bytes()
        if body.count(b'Version: 0.1.0~alpha.1\n') != 1:
            raise ValueError('fixture native metadata differs')
        control.write_bytes(body.replace(b'Version: 0.1.0~alpha.1\n', b'Version: 0.1.0~alpha.0\n'))
        execute(['dpkg-deb', '--root-owner-group', '-Zxz', '--build', str(work / 'deb'), str(deb)], 'deb-build')
        deb_after_headers = deb_headers(deb)
        if record['debian_tar_headers_before'] != deb_after_headers:
            raise ValueError('fixture Debian tar headers differ')
        stage = work / 'rpm-stage'
        stage.mkdir()
        for name, (body, mode) in cpio(bounded_command(['rpm2cpio', str(packages['fedora'])]), contents=True).items():
            target = stage / name
            target.parent.mkdir(parents=True, exist_ok=True)
            target.write_bytes(body)
            target.chmod(mode)
        shutil.copyfile(work / 'deb/DEBIAN/preinst', stage / 'candidate-preinst')
        execute(['bash', str(ROOT / 'packaging/build-rpm.sh'), str(out), str(stage), '--same-code-fixture-stage'], 'rpm-build')
        fixtures = {'debian': deb, 'fedora': out / 'fabrico11y-0.1.0-0.alpha.0.x86_64.rpm'}
        rpm_after = rpm_records(fixtures['fedora'])
        if rpm_before['metadata'] != rpm_after['metadata']:
            raise ValueError('fixture RPM stable native metadata differs')
        changed_scripts = {'debian': deb_controls(deb), 'fedora': rpm_after['scripts']}
        for family, package in fixtures.items():
            command = [str(package) if part == str(packages[family]) else part for part in version_commands[family]]
            version = bounded_command(command).decode().strip()
            actual = inventory(package)
            validate_fixture_transform(family, versions[family][0], version, before[family], actual,
                                       scripts[family], changed_scripts[family])
            record['packages'][family] = {'package': str(package), 'sha256': digest(package), 'version': version,
                                          'payload_identity': actual, 'script_identity': changed_scripts[family]}
        record['packages']['fedora']['native_metadata_identity'] = rpm_after['metadata']
        record['packages']['debian']['tar_header_identity'] = deb_after_headers
        if any(digest(Path(path)) != expected for path, expected in inputs.items()):
            raise ValueError('fixture canonical inputs changed during construction')
        validate_inputs(args.deb_receipt, args.rpm_receipt)
        validate_harness(args.harness_commit)
        record.update(state='built', exit=0, native_older={'debian': True, 'fedora': True})
    except (Exception, KeyboardInterrupt) as error:
        record['error'] = str(error) or type(error).__name__
        if (isinstance(error, (KeyboardInterrupt, subprocess.TimeoutExpired))
                or isinstance(error, subprocess.CalledProcessError) and error.returncode < 0):
            record['state'] = 'interrupted'
        if getattr(error, 'stderr', None) is not None:
            record['stderr'] = error.stderr.decode(errors='replace') if isinstance(error.stderr, bytes) else error.stderr
        if work is not None:
            record['failure_evidence'] = str(work)
            try:
                shutil.move(str(work), str(out / 'failed-fixture-work'))
                record['failure_evidence'] = str(out / 'failed-fixture-work')
            except Exception as archive_error:
                record['archive_error'] = str(archive_error)
    finally:
        if record['state'] == 'built' and work is not None and work.exists():
            try:
                shutil.rmtree(work)
            except OSError as cleanup_error:
                record.update(state='failed', exit=2, error=str(cleanup_error), failure_evidence=str(work))
        (out / 'fixture-receipt.json').write_text(json.dumps(record, indent=2) + '\n')
    print(json.dumps(record, indent=2))
    return record['exit']


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--deb-receipt', required=True, type=Path)
    parser.add_argument('--out', required=True, type=Path)
    parser.add_argument('--same-code-fixture', action='store_true')
    parser.add_argument('--rpm-receipt', type=Path)
    parser.add_argument('--frozen')
    parser.add_argument('--source-date-epoch', type=int)
    parser.add_argument('--harness-commit')
    args = parser.parse_args()
    require_limits()
    if args.same_code_fixture:
        return fixture_main(args)
    if any(value is not None for value in (args.rpm_receipt, args.frozen, args.source_date_epoch, args.harness_commit)):
        parser.error('fixture pins require --same-code-fixture')
    out = args.out.resolve()
    out.relative_to(STORAGE / 'results')
    out.mkdir(exist_ok=False, parents=True)
    source = json.loads(args.deb_receipt.read_text())
    package = Path(source['package'])
    with tempfile.TemporaryDirectory(dir=os.environ['FABRIC_SCRATCH_ROOT'], prefix='candidate-rpm-input-') as scratch:
        staged = Path(scratch) / 'validated'
        stage_candidate(package, args.deb_receipt, staged)
        provenance_hash = digest(staged / 'usr/share/doc/fabrico11y/CANDIDATE-PROVENANCE.json')
    epoch = source['source']['source_date_epoch']
    env = {**os.environ, 'SOURCE_DATE_EPOCH': str(epoch)}
    command = ['bash', str(ROOT / 'packaging/build-rpm.sh'), str(out), str(package), '--candidate-deb', str(args.deb_receipt)]
    record = {'state': 'failed', 'command': command, 'exit': 2,
              'deb_receipt': str(args.deb_receipt.resolve()), 'deb_receipt_sha256': digest(args.deb_receipt),
              'deb_sha256': source['package_sha256'], 'source': source['source'],
              'rpm_builder_sha256': digest(ROOT / 'packaging/build-rpm.sh'),
              'adapter_sha256': digest(Path(__file__)), 'package_family': 'fedora'}
    try:
        record['build_exit'] = run_build(command, env, out)
        if record['build_exit']:
            raise ValueError('RPM build failed')
        packages = list(out.glob('*.rpm'))
        if len(packages) != 1:
            raise ValueError('expected exactly one candidate RPM')
        payload = assert_equivalent(package, packages[0])
        if inventory(packages[0])['usr/share/doc/fabrico11y/CANDIDATE-PROVENANCE.json']['sha256'] != provenance_hash:
            raise ValueError('RPM candidate provenance differs from exact validated build inputs')
        record.update(state='built', package=str(packages[0]), package_sha256=digest(packages[0]),
                      payload_identity=payload, candidate_provenance_sha256=provenance_hash, exit=0)
    except (ValueError, OSError, subprocess.SubprocessError) as error:
        record.update(error=str(error), exit=2)
    (out / 'receipt.json').write_text(json.dumps(record, indent=2) + '\n')
    print(json.dumps(record, indent=2))
    return record['exit']


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

"""Bounded exact regular-file inventories for candidate Deb/RPM equivalence."""
import hashlib
import io
import os
from pathlib import PurePosixPath
import stat
import select
import subprocess
import tarfile
import time

CAP = 128 * 1024**2


def bounded_command(command):
    process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    deadline = time.monotonic() + 30
    result, errors = bytearray(), bytearray()
    streams = {process.stdout: result, process.stderr: errors}
    try:
        while streams:
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                raise subprocess.TimeoutExpired(command, 30)
            ready, _, _ = select.select(list(streams), [], [], min(1, remaining))
            for stream in ready:
                chunk = os.read(stream.fileno(), 65536)
                if not chunk:
                    del streams[stream]
                elif len(streams[stream]) + len(chunk) > CAP:
                    raise ValueError('payload extraction exceeds bounded inventory cap')
                else:
                    streams[stream].extend(chunk)
        status = process.wait(timeout=max(0.01, deadline - time.monotonic()))
        if status < 0:
            raise subprocess.CalledProcessError(status, command, output=bytes(result), stderr=bytes(errors))
        if status:
            raise ValueError('payload extraction failed (exit ' + str(status) + ')')
        return bytes(result)
    except ValueError as error:
        raise ValueError(str(error) + '; command=' + repr(command)
                         + '; stderr=' + errors.decode(errors='replace')) from error
    except subprocess.TimeoutExpired as error:
        error.output, error.stderr = bytes(result), bytes(errors)
        raise
    finally:
        if process.poll() is None:
            process.kill()
            process.wait(timeout=10)
        process.stdout.close()
        process.stderr.close()


def identity(body):
    return {'bytes': len(body), 'sha256': hashlib.sha256(body).hexdigest()}


def safe(name):
    path = PurePosixPath(name.removeprefix('./'))
    if path.is_absolute() or '..' in path.parts or not path.parts or path.parts[0] != 'usr':
        raise ValueError('unsafe candidate payload path: ' + name)
    return str(path)


def cpio(raw, contents=False):
    if len(raw) > CAP:
        raise ValueError('RPM payload exceeds bounded inventory cap')
    result = {}
    offset = 0
    while offset + 110 <= len(raw):
        if raw[offset:offset+6] != b'070701':
            raise ValueError('unsupported RPM cpio encoding')
        fields = [int(raw[offset+6+i*8:offset+14+i*8], 16) for i in range(13)]
        mode, size, namesize = fields[1], fields[6], fields[11]
        start = offset + 110
        name_raw = raw[start:start+namesize]
        if not namesize or len(name_raw) != namesize or name_raw[-1] != 0:
            raise ValueError('truncated RPM cpio name')
        name = name_raw[:-1].decode()
        content_at = (start + namesize + 3) & ~3
        body = raw[content_at:content_at+size]
        if len(body) != size:
            raise ValueError('truncated RPM cpio body')
        offset = (content_at + size + 3) & ~3
        if name == 'TRAILER!!!':
            return result
        if stat.S_ISDIR(mode):
            continue
        name = safe(name)
        if not stat.S_ISREG(mode) or name in result:
            raise ValueError('nonregular or duplicate RPM payload')
        result[name] = (body, stat.S_IMODE(mode)) if contents else identity(body)
    raise ValueError('RPM cpio trailer missing')


def inventory(package):
    if package.suffix == '.rpm':
        raw = bounded_command(['rpm2cpio', str(package)])
        return cpio(raw)
    raw = bounded_command(['ar', 'p', str(package), 'data.tar.xz'])
    if len(raw) > CAP:
        raise ValueError('Debian compressed payload exceeds cap')
    result = {}
    size = 0
    with tarfile.open(fileobj=io.BytesIO(raw)) as archive:
        for member in archive:
            if member.isdir():
                continue
            name = safe(member.name)
            size += member.size
            if not member.isfile() or member.size > 32 * 1024**2 or size > CAP or name in result:
                raise ValueError('unsafe Debian payload entry')
            result[name] = identity(archive.extractfile(member).read())
    return result


def assert_equivalent(deb, rpm):
    expected = inventory(deb)
    actual = inventory(rpm)
    extra = {'usr/share/doc/fabrico11y/CANDIDATE-PROVENANCE.json'}
    if set(actual) != set(expected) | extra or any(actual[name] != value for name, value in expected.items()):
        raise ValueError('RPM must preserve every exact Debian regular payload byte')
    return expected